fix: add scheduled job execution history (#199)
This commit is contained in:
QiuSW
@@ -2,16 +2,8 @@
|
||||
generated: true (请先修改 Gitea Wiki，禁止直接编辑本文件)
|
||||
wiki_page: Architecture-and-Code-Map
|
||||
wiki_url: https://git.ilapage.cn/OPC/goauto/wiki/Architecture-and-Code-Map.-
|
||||
wiki_revision: 568d40fd882dc520e247226c4dd4d5c6b03122d9
|
||||
synchronized_at: 2026-09-02T13:23:01Z
|
||||
<!-- gitea-wiki-mirror:end -->
|
||||
|
||||
<!-- gitea-wiki-mirror:start -->
|
||||
generated: true (请先修改 Gitea Wiki，禁止直接编辑本文件)
|
||||
wiki_page: Architecture-and-Code-Map
|
||||
wiki_url: https://git.ilapage.cn/OPC/goauto/wiki/Architecture-and-Code-Map.-
|
||||
wiki_revision: dc5ccd2104e8ded715c0c0bb23d3827dc9a7fd7a
|
||||
synchronized_at: 2026-08-29T01:39:35Z
|
||||
wiki_revision: a8ea0f6ae4c4a05b10ba047516fda23cfb188d13
|
||||
synchronized_at: 2026-09-02T15:05:31Z
|
||||
<!-- gitea-wiki-mirror:end -->
|
||||
|
||||
# 架构与代码地图
|
||||
@@ -333,3 +325,11 @@ PddProductDetailCollector
|
||||
- `syb_spec_ai_parse_run` 保存全局批次、活动槽、租约与结构化计数；`syb_spec_ai_parse_work_item` 按 `syb_product_id` 唯一保存输入指纹、尝试次数、冷却和逐行租约。
|
||||
- `syb_product` 的 `ai_confirmed`、置信度、限长理由、确认时间和隐藏输入指纹记录 AI 确认事实；`parse_status` 仍只记录确定性解析器结果，`manually_confirmed` 仍只代表人工决定。
|
||||
- Provider 只接收单条 `productSpec` 和关联蝦皮商品的颜色/尺码候选；返回值必须逐字属于对应候选。采购可信门禁接受仍有效的 AI 确认，但候选消失后立即 fail-closed。
|
||||
|
||||
## 定时任务持久化执行历史（#199）
|
||||
|
||||
- `server/app/jobs/execution_log.go` 为 Exec 与 HTTP 两类调度入口统一记录一次执行生命周期；`server/app/jobs/models/sys_job_execution_log.go` 对应表 `sys_job_execution_log`，保存执行标识、任务/调用目标快照、定时触发类型、开始/结束时间、耗时、`running` / `succeeded` / `failed` / `interrupted` 状态及脱敏错误码和摘要。
|
||||
- 服务启动时按当前“每个数据库一个调度器实例”的拓扑，把上次进程遗留的 `running` 记录安全收敛为 `interrupted`；任务删除后历史仍保留并可只读查询。
|
||||
- `GET /api/v1/sysjob/:id/execution-logs` 由 `server/app/jobs/service/execution_log.go`、`apis/execution_log.go` 和 `router/sys_job.go` 提供任务级分页、状态与开始时间过滤；沿用隐藏菜单 `JobLog` 的角色菜单绑定和精确 GET 权限。Web 入口为 `web/src/views/schedule/index.vue` 的单选“日志”按钮，详情页为 `web/src/views/schedule/log.vue`。
|
||||
- 执行历史明确不保存任务参数、AI Provider 地址或密钥、第三方原始响应和业务原始载荷；当前不提供删除、保留期限自动化、WebSocket 实时流或立即执行动作。
|
||||
- 追加迁移为 `server/cmd/migrate/migration/version-local/1788357000000_sys_job_execution_log.go`：创建执行历史表、登记只读 API、关联 `JobLog` 菜单，并只给迁移前已绑定该菜单的角色补充精确 Casbin 权限。
|
||||
|
||||
@@ -2,8 +2,8 @@
|
||||
generated: true (请先修改 Gitea Wiki，禁止直接编辑本文件)
|
||||
wiki_page: Deployment-and-Operations
|
||||
wiki_url: https://git.ilapage.cn/OPC/goauto/wiki/Deployment-and-Operations.-
|
||||
wiki_revision: 6b6458ea30bc2140ad39dbf0c28c0d5896ee7618
|
||||
synchronized_at: 2026-09-02T13:23:38Z
|
||||
wiki_revision: 16d97e7596dc287786f702c71e25172b75cc4d56
|
||||
synchronized_at: 2026-09-02T15:05:56Z
|
||||
<!-- gitea-wiki-mirror:end -->
|
||||
|
||||
# 部署与运维
|
||||
@@ -63,3 +63,10 @@ Provider 故障日志只允许记录调用关联 ID、操作类型、耗时、
|
||||
- 系统任务默认关闭，默认 Cron 为每小时第 5 分钟、参数 `{"batchLimit":20}`；迁移重跑不得覆盖管理员后续修改的 Cron、参数或启停状态。建议先运行该任务，再由默认第 15 分钟的 #195 完成蝦皮到 PDD 的规格匹配。
|
||||
- 多实例通过可空唯一活动槽和租约保证全局单批运行；逐行工作项按输入指纹去重。Provider 临时失败至少 60 分钟后重试且最多 3 次，低置信度/无结果在输入不变时不重复调用。
|
||||
- 上线顺序为：备份数据库、执行追加迁移、发布服务并保持任务关闭、核验表/字段/任务种子，再由管理员决定是否启用。排错只查看运行计数、工作状态和脱敏错误，不输出 API Key、Provider 原始响应、完整 RawJSON 或订单数据。
|
||||
|
||||
## 定时任务执行历史运维（#199）
|
||||
|
||||
- 发布 #199 时先备份数据库，再执行追加迁移 `1788357000000_sys_job_execution_log.go`，随后发布服务与 Web。迁移创建 `sys_job_execution_log`、登记 `GET /api/v1/sysjob/:id/execution-logs`，并只为原本绑定隐藏菜单 `JobLog` 的角色写入精确读取权限；菜单缺失时迁移明确失败，不扩大角色授权。
|
||||
- 服务启动会将上次进程遗留的 `running` 执行记录标记为 `interrupted`。该恢复依赖当前线上每个数据库只运行一个调度器实例；扩展为多调度器前必须另建工单引入实例租约，不能直接复用此判断。
|
||||
- 排错从 Admin 定时任务页单选任务后进入“日志”，按状态和开始时间查询。记录只含稳定错误码和脱敏摘要；需要定位细节时查看受控服务日志，不得把任务参数、Provider 配置/响应、密钥或业务原始数据复制进执行历史。
|
||||
- 当前没有执行历史删除接口和自动保留策略；删除定时任务不删除历史。数据库容量治理需要另建工单评估。#198 的 `GoAutoSYBSpecAIParse` 在 #199 发布和迁移后仍保持关闭，启用必须由管理员另行确认。
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
package apis
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
|
||||
jobservice "go-admin/app/jobs/service"
|
||||
)
|
||||
|
||||
func (e SysJob) ListExecutionLogs(c *gin.Context) {
|
||||
jobID, err := strconv.Atoi(c.Param("id"))
|
||||
if err != nil || jobID < 1 {
|
||||
c.JSON(http.StatusUnprocessableEntity, gin.H{"code": 400, "msg": "jobId 无效"})
|
||||
return
|
||||
}
|
||||
page, err := positiveQueryInt(c.Query("pageIndex"), 1)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusUnprocessableEntity, gin.H{"code": 400, "msg": "pageIndex 必须是正整数"})
|
||||
return
|
||||
}
|
||||
pageSize, err := positiveQueryInt(c.Query("pageSize"), 20)
|
||||
if err != nil || pageSize > 100 {
|
||||
c.JSON(http.StatusUnprocessableEntity, gin.H{"code": 400, "msg": "pageSize 必须是 1 到 100 的整数"})
|
||||
return
|
||||
}
|
||||
startedFrom, err := optionalTime(c.Query("startedFrom"))
|
||||
if err != nil {
|
||||
c.JSON(http.StatusUnprocessableEntity, gin.H{"code": 400, "msg": "startedFrom 必须是 RFC3339 时间"})
|
||||
return
|
||||
}
|
||||
startedTo, err := optionalTime(c.Query("startedTo"))
|
||||
if err != nil {
|
||||
c.JSON(http.StatusUnprocessableEntity, gin.H{"code": 400, "msg": "startedTo 必须是 RFC3339 时间"})
|
||||
return
|
||||
}
|
||||
|
||||
e.MakeContext(c)
|
||||
db, err := e.GetOrm()
|
||||
if err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"code": 500, "msg": "服务端处理失败"})
|
||||
return
|
||||
}
|
||||
response, err := jobservice.NewExecutionLogService(db).List(c.Request.Context(), jobID, jobservice.ExecutionLogListRequest{
|
||||
Page: page, PageSize: pageSize, Status: strings.TrimSpace(c.Query("status")),
|
||||
StartedFrom: startedFrom, StartedTo: startedTo,
|
||||
})
|
||||
if err != nil {
|
||||
switch {
|
||||
case errors.Is(err, jobservice.ErrExecutionLogInvalidRequest):
|
||||
c.JSON(http.StatusUnprocessableEntity, gin.H{"code": 400, "msg": err.Error()})
|
||||
case errors.Is(err, jobservice.ErrExecutionLogJobNotFound):
|
||||
c.JSON(http.StatusNotFound, gin.H{"code": 404, "msg": "定时任务不存在"})
|
||||
default:
|
||||
e.GetLogger().Errorf("list scheduled job execution logs failed job_id=%d: %v", jobID, err)
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"code": 500, "msg": "服务端处理失败"})
|
||||
}
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, gin.H{"code": 200, "data": response})
|
||||
}
|
||||
|
||||
func positiveQueryInt(value string, fallback int) (int, error) {
|
||||
if strings.TrimSpace(value) == "" {
|
||||
return fallback, nil
|
||||
}
|
||||
parsed, err := strconv.Atoi(value)
|
||||
if err != nil || parsed < 1 {
|
||||
return 0, errors.New("invalid positive integer")
|
||||
}
|
||||
return parsed, nil
|
||||
}
|
||||
|
||||
func optionalTime(value string) (*time.Time, error) {
|
||||
if strings.TrimSpace(value) == "" {
|
||||
return nil, nil
|
||||
}
|
||||
parsed, err := time.Parse(time.RFC3339, value)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &parsed, nil
|
||||
}
|
||||
@@ -0,0 +1,132 @@
|
||||
package jobs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
log "github.com/go-admin-team/go-admin-core/logger"
|
||||
"github.com/google/uuid"
|
||||
"gorm.io/gorm"
|
||||
|
||||
"go-admin/app/jobs/models"
|
||||
)
|
||||
|
||||
const (
|
||||
executionErrorTargetMissing = "JOB_TARGET_NOT_FOUND"
|
||||
executionErrorExecFailed = "JOB_EXECUTION_FAILED"
|
||||
executionErrorHTTPFailed = "JOB_HTTP_FAILED"
|
||||
executionErrorPanicked = "JOB_EXECUTION_PANICKED"
|
||||
executionErrorInterrupted = "JOB_INTERRUPTED"
|
||||
)
|
||||
|
||||
type executionFailure struct {
|
||||
code string
|
||||
message string
|
||||
cause error
|
||||
}
|
||||
|
||||
func (failure *executionFailure) Error() string {
|
||||
if failure.cause != nil {
|
||||
return failure.cause.Error()
|
||||
}
|
||||
return failure.message
|
||||
}
|
||||
|
||||
func newExecutionFailure(code, message string, cause error) error {
|
||||
return &executionFailure{code: code, message: message, cause: cause}
|
||||
}
|
||||
|
||||
func runWithExecutionLog(db *gorm.DB, core JobCore, execute func() error) (executionErr error) {
|
||||
startedAt := time.Now().UTC()
|
||||
record := &models.SysJobExecutionLog{
|
||||
ExecutionID: uuid.NewString(), JobID: core.JobId,
|
||||
JobNameSnapshot: core.Name, InvokeTargetSnapshot: core.InvokeTarget,
|
||||
TriggerType: models.JobTriggerScheduled, Status: models.JobExecutionRunning,
|
||||
StartedAt: startedAt,
|
||||
}
|
||||
created := false
|
||||
if db != nil {
|
||||
if err := db.WithContext(context.Background()).Create(record).Error; err != nil {
|
||||
log.Errorf("[Job] execution log create failed job_id=%d: %v", core.JobId, err)
|
||||
} else {
|
||||
created = true
|
||||
}
|
||||
}
|
||||
|
||||
defer func() {
|
||||
if recovered := recover(); recovered != nil {
|
||||
executionErr = newExecutionFailure(executionErrorPanicked, "任务执行异常中断", nil)
|
||||
if created {
|
||||
finishExecutionLog(db, core.JobId, record, startedAt, executionErr)
|
||||
}
|
||||
return
|
||||
}
|
||||
if created {
|
||||
finishExecutionLog(db, core.JobId, record, startedAt, executionErr)
|
||||
}
|
||||
}()
|
||||
executionErr = execute()
|
||||
return executionErr
|
||||
}
|
||||
|
||||
func finishExecutionLog(db *gorm.DB, jobID int, record *models.SysJobExecutionLog, startedAt time.Time, executionErr error) {
|
||||
finishedAt := time.Now().UTC()
|
||||
updates := map[string]any{
|
||||
"status": models.JobExecutionSucceeded, "finished_at": finishedAt,
|
||||
"duration_ms": finishedAt.Sub(startedAt).Milliseconds(), "error_code": "", "error_message": "",
|
||||
}
|
||||
if executionErr != nil {
|
||||
code, message := publicExecutionFailure(executionErr)
|
||||
updates["status"] = models.JobExecutionFailed
|
||||
updates["error_code"] = code
|
||||
updates["error_message"] = message
|
||||
}
|
||||
if err := db.WithContext(context.Background()).Model(&models.SysJobExecutionLog{}).
|
||||
Where("id = ? AND status = ?", record.ID, models.JobExecutionRunning).Updates(updates).Error; err != nil {
|
||||
log.Errorf("[Job] execution log finish failed job_id=%d execution_id=%s: %v", jobID, record.ExecutionID, err)
|
||||
}
|
||||
}
|
||||
|
||||
func publicExecutionFailure(err error) (string, string) {
|
||||
var failure *executionFailure
|
||||
if errors.As(err, &failure) {
|
||||
return failure.code, failure.message
|
||||
}
|
||||
return executionErrorExecFailed, "任务执行失败，请查看受控服务日志"
|
||||
}
|
||||
|
||||
// RecoverInterruptedExecutionLogs closes invocations left running by the
|
||||
// previous process. Production currently runs one scheduler per database.
|
||||
func RecoverInterruptedExecutionLogs(db *gorm.DB) error {
|
||||
if db == nil {
|
||||
return nil
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
var records []models.SysJobExecutionLog
|
||||
if err := db.WithContext(context.Background()).Where("status = ?", models.JobExecutionRunning).Find(&records).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
return db.WithContext(context.Background()).Transaction(func(tx *gorm.DB) error {
|
||||
for _, record := range records {
|
||||
duration := now.Sub(record.StartedAt).Milliseconds()
|
||||
if duration < 0 {
|
||||
duration = 0
|
||||
}
|
||||
if err := tx.Model(&models.SysJobExecutionLog{}).
|
||||
Where("id = ? AND status = ?", record.ID, models.JobExecutionRunning).Updates(map[string]any{
|
||||
"status": models.JobExecutionInterrupted, "finished_at": now,
|
||||
"duration_ms": duration, "error_code": executionErrorInterrupted,
|
||||
"error_message": "服务重启前任务未完成",
|
||||
}).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
func missingExecutionTarget(target string) error {
|
||||
return newExecutionFailure(executionErrorTargetMissing, "任务调用目标未注册", fmt.Errorf("job target %q is not registered", target))
|
||||
}
|
||||
@@ -0,0 +1,113 @@
|
||||
package jobs
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"gorm.io/driver/sqlite"
|
||||
"gorm.io/gorm"
|
||||
|
||||
"go-admin/app/jobs/models"
|
||||
)
|
||||
|
||||
func jobExecutionTestDB(t *testing.T) *gorm.DB {
|
||||
t.Helper()
|
||||
db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.AutoMigrate(&models.SysJobExecutionLog{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return db
|
||||
}
|
||||
|
||||
func TestRunWithExecutionLogRecordsSuccessAndSanitizedFailure(t *testing.T) {
|
||||
db := jobExecutionTestDB(t)
|
||||
core := JobCore{JobId: 7, Name: "测试任务", InvokeTarget: "TestTarget"}
|
||||
if err := runWithExecutionLog(db, core, func() error { return nil }); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
rawSecret := "token=secret-value productSpec=private"
|
||||
if err := runWithExecutionLog(db, core, func() error { return errors.New(rawSecret) }); err == nil {
|
||||
t.Fatal("failed execution must return its error")
|
||||
}
|
||||
var records []models.SysJobExecutionLog
|
||||
if err := db.Order("id asc").Find(&records).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(records) != 2 || records[0].Status != models.JobExecutionSucceeded || records[1].Status != models.JobExecutionFailed {
|
||||
t.Fatalf("unexpected records: %+v", records)
|
||||
}
|
||||
if records[1].ErrorCode != executionErrorExecFailed || records[1].ErrorMessage == rawSecret || records[1].ErrorMessage == "" {
|
||||
t.Fatalf("failure was not safely summarized: %+v", records[1])
|
||||
}
|
||||
if records[0].FinishedAt == nil || records[1].FinishedAt == nil || records[0].ExecutionID == records[1].ExecutionID {
|
||||
t.Fatal("execution lifecycle or unique IDs were not recorded")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRecoverInterruptedExecutionLogs(t *testing.T) {
|
||||
db := jobExecutionTestDB(t)
|
||||
started := time.Now().UTC().Add(-2 * time.Second)
|
||||
record := models.SysJobExecutionLog{
|
||||
ExecutionID: "00000000-0000-4000-8000-000000000010", JobID: 9,
|
||||
JobNameSnapshot: "中断任务", InvokeTargetSnapshot: "Interrupted",
|
||||
TriggerType: models.JobTriggerScheduled, Status: models.JobExecutionRunning, StartedAt: started,
|
||||
}
|
||||
if err := db.Create(&record).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := RecoverInterruptedExecutionLogs(db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.First(&record, record.ID).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if record.Status != models.JobExecutionInterrupted || record.FinishedAt == nil || record.ErrorCode != executionErrorInterrupted || record.DurationMS < 1000 {
|
||||
t.Fatalf("record was not safely interrupted: %+v", record)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMissingExecutionTargetHasPublicSafeMessage(t *testing.T) {
|
||||
code, message := publicExecutionFailure(missingExecutionTarget("SecretTarget"))
|
||||
if code != executionErrorTargetMissing || message != "任务调用目标未注册" {
|
||||
t.Fatalf("unexpected target failure %s %s", code, message)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExecJobRecordsMissingTargetAsFailure(t *testing.T) {
|
||||
db := jobExecutionTestDB(t)
|
||||
previousJobList := jobList
|
||||
jobList = map[string]JobExec{}
|
||||
t.Cleanup(func() { jobList = previousJobList })
|
||||
|
||||
job := &ExecJob{JobCore: JobCore{JobId: 10, Name: "未注册任务", InvokeTarget: "MissingTarget"}, DB: db}
|
||||
job.Run()
|
||||
|
||||
var record models.SysJobExecutionLog
|
||||
if err := db.First(&record).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if record.Status != models.JobExecutionFailed || record.ErrorCode != executionErrorTargetMissing || record.ErrorMessage != "任务调用目标未注册" {
|
||||
t.Fatalf("missing target was not safely recorded: %+v", record)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunWithExecutionLogSafelyRecordsPanic(t *testing.T) {
|
||||
db := jobExecutionTestDB(t)
|
||||
err := runWithExecutionLog(db, JobCore{JobId: 11, Name: "异常任务", InvokeTarget: "Panic"}, func() error {
|
||||
panic("provider-secret")
|
||||
})
|
||||
if code, message := publicExecutionFailure(err); code != executionErrorPanicked || message != "任务执行异常中断" {
|
||||
t.Fatalf("panic was not returned as a safe failure: %s %s", code, message)
|
||||
}
|
||||
var record models.SysJobExecutionLog
|
||||
if err := db.First(&record).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if record.Status != models.JobExecutionFailed || record.ErrorCode != executionErrorPanicked || record.ErrorMessage != "任务执行异常中断" {
|
||||
t.Fatalf("panic was not safely persisted: %+v", record)
|
||||
}
|
||||
}
|
||||
+33
-22
@@ -33,6 +33,7 @@ type JobCore struct {
|
||||
// HttpJob 任务类型 http
|
||||
type HttpJob struct {
|
||||
JobCore
|
||||
DB *gorm.DB
|
||||
}
|
||||
|
||||
type ExecJob struct {
|
||||
@@ -42,14 +43,16 @@ type ExecJob struct {
|
||||
|
||||
func (e *ExecJob) Run() {
|
||||
startTime := time.Now()
|
||||
var obj = jobList[e.InvokeTarget]
|
||||
if obj == nil {
|
||||
log.Warn("[Job] ExecJob Run job nil")
|
||||
return
|
||||
}
|
||||
err := CallExecWithDB(obj.(JobExec), e.DB, e.Args)
|
||||
err := runWithExecutionLog(e.DB, e.JobCore, func() error {
|
||||
obj := jobList[e.InvokeTarget]
|
||||
if obj == nil {
|
||||
return missingExecutionTarget(e.InvokeTarget)
|
||||
}
|
||||
return CallExecWithDB(obj, e.DB, e.Args)
|
||||
})
|
||||
if err != nil {
|
||||
log.Errorf("[Job] JobCore %s failed: %v", e.Name, err)
|
||||
code, _ := publicExecutionFailure(err)
|
||||
log.Errorf("[Job] JobCore %s failed error_code=%s", e.Name, code)
|
||||
return
|
||||
}
|
||||
// 结束时间
|
||||
@@ -68,22 +71,26 @@ func (e *ExecJob) Run() {
|
||||
func (h *HttpJob) Run() {
|
||||
|
||||
startTime := time.Now()
|
||||
var count = 0
|
||||
var err error
|
||||
var str string
|
||||
/* 循环 */
|
||||
LOOP:
|
||||
if count < retryCount {
|
||||
/* 跳过迭代 */
|
||||
str, err = pkg.Get(h.InvokeTarget)
|
||||
if err != nil {
|
||||
// 如果失败暂停一段时间重试
|
||||
log.Warnf("[Job] mission failed! %v", err)
|
||||
log.Warnf("[Job] Retry after the task fails %d seconds! %s \n", (count+1)*5, str)
|
||||
time.Sleep(time.Duration(count+1) * 5 * time.Second)
|
||||
count = count + 1
|
||||
goto LOOP
|
||||
err := runWithExecutionLog(h.DB, h.JobCore, func() error {
|
||||
var lastErr error
|
||||
for count := 0; count < retryCount; count++ {
|
||||
_, requestErr := pkg.Get(h.InvokeTarget)
|
||||
if requestErr == nil {
|
||||
return nil
|
||||
}
|
||||
lastErr = requestErr
|
||||
log.Warnf("[Job] HTTP mission failed attempt=%d", count+1)
|
||||
if count+1 < retryCount {
|
||||
log.Warnf("[Job] Retry after the task fails %d seconds!\n", (count+1)*5)
|
||||
time.Sleep(time.Duration(count+1) * 5 * time.Second)
|
||||
}
|
||||
}
|
||||
return newExecutionFailure(executionErrorHTTPFailed, "HTTP 任务请求失败", lastErr)
|
||||
})
|
||||
if err != nil {
|
||||
code, _ := publicExecutionFailure(err)
|
||||
log.Errorf("[Job] JobCore %s failed error_code=%s", h.Name, code)
|
||||
return
|
||||
}
|
||||
// 结束时间
|
||||
endTime := time.Now()
|
||||
@@ -102,6 +109,9 @@ func Setup(dbs map[string]*gorm.DB) {
|
||||
fmt.Println(time.Now().Format(timeFormat), " [INFO] JobCore Starting...")
|
||||
|
||||
for k, db := range dbs {
|
||||
if err := RecoverInterruptedExecutionLogs(db); err != nil {
|
||||
log.Errorf("[Job] recover interrupted execution logs failed: %v", err)
|
||||
}
|
||||
sdk.Runtime.SetCrontab(k, cronjob.NewWithSeconds())
|
||||
setup(k, db)
|
||||
}
|
||||
@@ -127,6 +137,7 @@ func setup(key string, db *gorm.DB) {
|
||||
for i := 0; i < len(jobList); i++ {
|
||||
if jobList[i].JobType == 1 {
|
||||
j := &HttpJob{}
|
||||
j.DB = db
|
||||
j.InvokeTarget = jobList[i].InvokeTarget
|
||||
j.CronExpression = jobList[i].CronExpression
|
||||
j.JobId = jobList[i].JobId
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
package models
|
||||
|
||||
import "time"
|
||||
|
||||
const (
|
||||
JobExecutionRunning = "running"
|
||||
JobExecutionSucceeded = "succeeded"
|
||||
JobExecutionFailed = "failed"
|
||||
JobExecutionInterrupted = "interrupted"
|
||||
JobTriggerScheduled = "scheduled"
|
||||
)
|
||||
|
||||
// SysJobExecutionLog stores one scheduler invocation. Job arguments and raw
|
||||
// provider responses are deliberately excluded from this audit record.
|
||||
type SysJobExecutionLog struct {
|
||||
ID uint64 `json:"id" gorm:"primaryKey;autoIncrement"`
|
||||
ExecutionID string `json:"executionId" gorm:"size:36;not null;uniqueIndex:ux_sys_job_execution_id"`
|
||||
JobID int `json:"jobId" gorm:"not null;index:idx_sys_job_execution_job_started,priority:1"`
|
||||
JobNameSnapshot string `json:"jobName" gorm:"size:255;not null"`
|
||||
InvokeTargetSnapshot string `json:"invokeTarget" gorm:"size:255;not null"`
|
||||
TriggerType string `json:"triggerType" gorm:"size:16;not null"`
|
||||
Status string `json:"status" gorm:"size:16;not null;index:idx_sys_job_execution_status_started,priority:1"`
|
||||
StartedAt time.Time `json:"startedAt" gorm:"not null;index:idx_sys_job_execution_job_started,priority:2;index:idx_sys_job_execution_status_started,priority:2"`
|
||||
FinishedAt *time.Time `json:"finishedAt,omitempty"`
|
||||
DurationMS int64 `json:"durationMs" gorm:"not null;default:0"`
|
||||
ErrorCode string `json:"errorCode,omitempty" gorm:"size:64;not null;default:''"`
|
||||
ErrorMessage string `json:"errorMessage,omitempty" gorm:"size:500;not null;default:''"`
|
||||
CreatedAt time.Time `json:"createdAt"`
|
||||
UpdatedAt time.Time `json:"updatedAt"`
|
||||
}
|
||||
|
||||
func (SysJobExecutionLog) TableName() string { return "sys_job_execution_log" }
|
||||
@@ -24,6 +24,8 @@ func registerSysJobRouter(v1 *gin.RouterGroup, authMiddleware *jwt.GinJWTMiddlew
|
||||
list := make([]models2.SysJob, 0)
|
||||
return &list
|
||||
}))
|
||||
jobAPI := apis.SysJob{}
|
||||
r.GET("/:id/execution-logs", actions.PermissionAction(), jobAPI.ListExecutionLogs)
|
||||
r.GET("/:id", actions.PermissionAction(), actions.ViewAction(new(dto2.SysJobById), func() interface{} {
|
||||
return &dto2.SysJobItem{}
|
||||
}))
|
||||
|
||||
@@ -0,0 +1,107 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
|
||||
"go-admin/app/jobs/models"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrExecutionLogInvalidRequest = errors.New("invalid execution log request")
|
||||
ErrExecutionLogJobNotFound = errors.New("scheduled job not found")
|
||||
)
|
||||
|
||||
type ExecutionLogListRequest struct {
|
||||
Page int
|
||||
PageSize int
|
||||
Status string
|
||||
StartedFrom *time.Time
|
||||
StartedTo *time.Time
|
||||
}
|
||||
|
||||
type ExecutionLogJob struct {
|
||||
JobID int `json:"jobId"`
|
||||
JobName string `json:"jobName"`
|
||||
InvokeTarget string `json:"invokeTarget"`
|
||||
Deleted bool `json:"deleted"`
|
||||
}
|
||||
|
||||
type ExecutionLogListResponse struct {
|
||||
Job ExecutionLogJob `json:"job"`
|
||||
Items []models.SysJobExecutionLog `json:"items"`
|
||||
Total int64 `json:"total"`
|
||||
Page int `json:"page"`
|
||||
PageSize int `json:"pageSize"`
|
||||
}
|
||||
|
||||
type ExecutionLogService struct{ db *gorm.DB }
|
||||
|
||||
func NewExecutionLogService(db *gorm.DB) *ExecutionLogService {
|
||||
return &ExecutionLogService{db: db}
|
||||
}
|
||||
|
||||
func (service *ExecutionLogService) List(ctx context.Context, jobID int, request ExecutionLogListRequest) (ExecutionLogListResponse, error) {
|
||||
if service.db == nil || jobID < 1 {
|
||||
return ExecutionLogListResponse{}, fmt.Errorf("%w: jobId 无效", ErrExecutionLogInvalidRequest)
|
||||
}
|
||||
if request.Page < 1 {
|
||||
request.Page = 1
|
||||
}
|
||||
if request.PageSize < 1 {
|
||||
request.PageSize = 20
|
||||
}
|
||||
if request.PageSize > 100 {
|
||||
return ExecutionLogListResponse{}, fmt.Errorf("%w: pageSize 必须是 1 到 100 的整数", ErrExecutionLogInvalidRequest)
|
||||
}
|
||||
if request.Status != "" && !validExecutionStatus(request.Status) {
|
||||
return ExecutionLogListResponse{}, fmt.Errorf("%w: status 无效", ErrExecutionLogInvalidRequest)
|
||||
}
|
||||
if request.StartedFrom != nil && request.StartedTo != nil && request.StartedFrom.After(*request.StartedTo) {
|
||||
return ExecutionLogListResponse{}, fmt.Errorf("%w: 开始时间范围无效", ErrExecutionLogInvalidRequest)
|
||||
}
|
||||
|
||||
var job models.SysJob
|
||||
if err := service.db.WithContext(ctx).Unscoped().First(&job, jobID).Error; err != nil {
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return ExecutionLogListResponse{}, ErrExecutionLogJobNotFound
|
||||
}
|
||||
return ExecutionLogListResponse{}, err
|
||||
}
|
||||
|
||||
query := service.db.WithContext(ctx).Model(&models.SysJobExecutionLog{}).Where("job_id = ?", jobID)
|
||||
if request.Status != "" {
|
||||
query = query.Where("status = ?", request.Status)
|
||||
}
|
||||
if request.StartedFrom != nil {
|
||||
query = query.Where("started_at >= ?", request.StartedFrom.UTC())
|
||||
}
|
||||
if request.StartedTo != nil {
|
||||
query = query.Where("started_at <= ?", request.StartedTo.UTC())
|
||||
}
|
||||
var total int64
|
||||
if err := query.Count(&total).Error; err != nil {
|
||||
return ExecutionLogListResponse{}, err
|
||||
}
|
||||
items := make([]models.SysJobExecutionLog, 0, request.PageSize)
|
||||
if err := query.Order("started_at DESC, id DESC").Offset((request.Page - 1) * request.PageSize).Limit(request.PageSize).Find(&items).Error; err != nil {
|
||||
return ExecutionLogListResponse{}, err
|
||||
}
|
||||
return ExecutionLogListResponse{
|
||||
Job: ExecutionLogJob{JobID: job.JobId, JobName: job.JobName, InvokeTarget: job.InvokeTarget, Deleted: job.DeletedAt.Valid},
|
||||
Items: items, Total: total, Page: request.Page, PageSize: request.PageSize,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func validExecutionStatus(status string) bool {
|
||||
switch status {
|
||||
case models.JobExecutionRunning, models.JobExecutionSucceeded, models.JobExecutionFailed, models.JobExecutionInterrupted:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,87 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"gorm.io/driver/sqlite"
|
||||
"gorm.io/gorm"
|
||||
|
||||
"go-admin/app/jobs/models"
|
||||
)
|
||||
|
||||
func executionLogTestDB(t *testing.T) *gorm.DB {
|
||||
t.Helper()
|
||||
db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.AutoMigrate(&models.SysJob{}, &models.SysJobExecutionLog{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return db
|
||||
}
|
||||
|
||||
func TestExecutionLogListFiltersPaginatesAndKeepsDeletedJob(t *testing.T) {
|
||||
db := executionLogTestDB(t)
|
||||
job := models.SysJob{JobName: "测试任务", InvokeTarget: "TestTarget"}
|
||||
if err := db.Create(&job).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
base := time.Date(2026, 9, 2, 10, 0, 0, 0, time.UTC)
|
||||
records := []models.SysJobExecutionLog{
|
||||
{ExecutionID: "00000000-0000-4000-8000-000000000001", JobID: job.JobId, JobNameSnapshot: job.JobName, InvokeTargetSnapshot: job.InvokeTarget, TriggerType: models.JobTriggerScheduled, Status: models.JobExecutionSucceeded, StartedAt: base},
|
||||
{ExecutionID: "00000000-0000-4000-8000-000000000002", JobID: job.JobId, JobNameSnapshot: job.JobName, InvokeTargetSnapshot: job.InvokeTarget, TriggerType: models.JobTriggerScheduled, Status: models.JobExecutionFailed, StartedAt: base.Add(time.Hour)},
|
||||
{ExecutionID: "00000000-0000-4000-8000-000000000003", JobID: job.JobId, JobNameSnapshot: job.JobName, InvokeTargetSnapshot: job.InvokeTarget, TriggerType: models.JobTriggerScheduled, Status: models.JobExecutionFailed, StartedAt: base.Add(2 * time.Hour)},
|
||||
}
|
||||
if err := db.Create(&records).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
from, to := base.Add(30*time.Minute), base.Add(3*time.Hour)
|
||||
result, err := NewExecutionLogService(db).List(context.Background(), job.JobId, ExecutionLogListRequest{
|
||||
Page: 1, PageSize: 1, Status: models.JobExecutionFailed, StartedFrom: &from, StartedTo: &to,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if result.Total != 2 || len(result.Items) != 1 || result.Items[0].ExecutionID != records[2].ExecutionID {
|
||||
t.Fatalf("unexpected filtered page: %+v", result)
|
||||
}
|
||||
if err := db.Delete(&job).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
deleted, err := NewExecutionLogService(db).List(context.Background(), job.JobId, ExecutionLogListRequest{Page: 1, PageSize: 20})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !deleted.Job.Deleted || deleted.Job.JobName != job.JobName || deleted.Total != 3 {
|
||||
t.Fatalf("deleted job history unavailable: %+v", deleted)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExecutionLogListValidationAndNotFound(t *testing.T) {
|
||||
db := executionLogTestDB(t)
|
||||
service := NewExecutionLogService(db)
|
||||
if _, err := service.List(context.Background(), 0, ExecutionLogListRequest{}); !errors.Is(err, ErrExecutionLogInvalidRequest) {
|
||||
t.Fatalf("invalid job id error = %v", err)
|
||||
}
|
||||
if _, err := service.List(context.Background(), 1, ExecutionLogListRequest{PageSize: 101}); !errors.Is(err, ErrExecutionLogInvalidRequest) {
|
||||
t.Fatalf("invalid page size error = %v", err)
|
||||
}
|
||||
job := models.SysJob{JobName: "测试任务", InvokeTarget: "TestTarget"}
|
||||
if err := db.Create(&job).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := service.List(context.Background(), job.JobId, ExecutionLogListRequest{Status: "unknown"}); !errors.Is(err, ErrExecutionLogInvalidRequest) {
|
||||
t.Fatalf("invalid status error = %v", err)
|
||||
}
|
||||
from, to := time.Now(), time.Now().Add(-time.Hour)
|
||||
if _, err := service.List(context.Background(), job.JobId, ExecutionLogListRequest{StartedFrom: &from, StartedTo: &to}); !errors.Is(err, ErrExecutionLogInvalidRequest) {
|
||||
t.Fatalf("invalid range error = %v", err)
|
||||
}
|
||||
if _, err := service.List(context.Background(), 999, ExecutionLogListRequest{}); !errors.Is(err, ErrExecutionLogJobNotFound) {
|
||||
t.Fatalf("not found error = %v", err)
|
||||
}
|
||||
}
|
||||
@@ -62,6 +62,7 @@ func (e *SysJob) StartJob(c *dto.GeneralGetDto) error {
|
||||
|
||||
if data.JobType == 1 {
|
||||
var j = &jobs.HttpJob{}
|
||||
j.DB = e.Orm.WithContext(context.Background())
|
||||
j.InvokeTarget = data.InvokeTarget
|
||||
j.CronExpression = data.CronExpression
|
||||
j.JobId = data.JobId
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
package version_local
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"runtime"
|
||||
|
||||
jobsmodels "go-admin/app/jobs/models"
|
||||
"go-admin/cmd/migrate/migration"
|
||||
migrationmodels "go-admin/cmd/migrate/migration/models"
|
||||
common "go-admin/common/models"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
const jobExecutionLogsAPIPath = "/api/v1/sysjob/:id/execution-logs"
|
||||
|
||||
func init() {
|
||||
_, fileName, _, _ := runtime.Caller(0)
|
||||
migration.Migrate.SetVersion(migration.GetFilename(fileName), migrateSysJobExecutionLog)
|
||||
}
|
||||
|
||||
func migrateSysJobExecutionLog(db *gorm.DB, version string) error {
|
||||
return db.Transaction(func(tx *gorm.DB) error {
|
||||
if err := ensureSysJobExecutionLog(tx); err != nil {
|
||||
return err
|
||||
}
|
||||
return tx.Create(&common.Migration{Version: version}).Error
|
||||
})
|
||||
}
|
||||
|
||||
func ensureSysJobExecutionLog(db *gorm.DB) error {
|
||||
if err := db.AutoMigrate(&jobsmodels.SysJobExecutionLog{}); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var logMenu migrationmodels.SysMenu
|
||||
if err := db.Where("menu_name = ?", "JobLog").First(&logMenu).Error; err != nil {
|
||||
return fmt.Errorf("find JobLog menu: %w", err)
|
||||
}
|
||||
api := migrationmodels.SysApi{}
|
||||
if err := db.Where(migrationmodels.SysApi{Path: jobExecutionLogsAPIPath, Action: "GET"}).
|
||||
Attrs(migrationmodels.SysApi{Title: "查询定时任务执行日志", Type: "BUS"}).FirstOrCreate(&api).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if err := db.Model(&logMenu).Association("SysApi").Append(&api); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var roleIDs []int
|
||||
if err := db.Table("sys_role_menu").Where("menu_id = ?", logMenu.MenuId).Distinct().Pluck("role_id", &roleIDs).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if len(roleIDs) == 0 {
|
||||
return nil
|
||||
}
|
||||
var roles []migrationmodels.SysRole
|
||||
if err := db.Where("role_id IN ?", roleIDs).Find(&roles).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
for _, role := range roles {
|
||||
rule := purchaserCasbinRule{Ptype: "p", V0: role.RoleKey, V1: jobExecutionLogsAPIPath, V2: "GET"}
|
||||
if err := db.Where("ptype = ? AND v0 = ? AND v1 = ? AND v2 = ?", rule.Ptype, rule.V0, rule.V1, rule.V2).
|
||||
FirstOrCreate(&rule).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
+78
@@ -0,0 +1,78 @@
|
||||
package version_local
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
jobsmodels "go-admin/app/jobs/models"
|
||||
migrationmodels "go-admin/cmd/migrate/migration/models"
|
||||
|
||||
"gorm.io/driver/sqlite"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
func TestEnsureSysJobExecutionLogIsIdempotentAndGrantsOnlyBoundRoles(t *testing.T) {
|
||||
db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.AutoMigrate(
|
||||
&migrationmodels.SysMenu{}, &migrationmodels.SysApi{}, &migrationmodels.SysRole{}, &purchaserCasbinRule{},
|
||||
); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
menu := migrationmodels.SysMenu{MenuName: "JobLog", Title: "日志", Path: "/schedule/log", Component: "/schedule/log"}
|
||||
if err := db.Create(&menu).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
bound := migrationmodels.SysRole{RoleName: "日志管理员", RoleKey: "job_logger"}
|
||||
unbound := migrationmodels.SysRole{RoleName: "无日志权限", RoleKey: "no_job_logs"}
|
||||
if err := db.Create(&bound).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Create(&unbound).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Model(&bound).Association("SysMenu").Append(&menu); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
custom := purchaserCasbinRule{Ptype: "p", V0: bound.RoleKey, V1: "/custom", V2: "GET"}
|
||||
if err := db.Create(&custom).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if err := ensureSysJobExecutionLog(db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := ensureSysJobExecutionLog(db); err != nil {
|
||||
t.Fatalf("migration helper must be repeatable: %v", err)
|
||||
}
|
||||
if !db.Migrator().HasTable(&jobsmodels.SysJobExecutionLog{}) {
|
||||
t.Fatal("execution log table was not created")
|
||||
}
|
||||
var apiCount, menuAPI, boundPolicy, unboundPolicy, customPolicy int64
|
||||
db.Model(&migrationmodels.SysApi{}).Where("path = ? AND action = ?", jobExecutionLogsAPIPath, "GET").Count(&apiCount)
|
||||
var api migrationmodels.SysApi
|
||||
if err := db.Where("path = ? AND action = ?", jobExecutionLogsAPIPath, "GET").First(&api).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
db.Table("sys_menu_api_rule").Where("sys_menu_menu_id = ? AND sys_api_id = ?", menu.MenuId, api.Id).Count(&menuAPI)
|
||||
db.Model(&purchaserCasbinRule{}).Where("v0 = ? AND v1 = ? AND v2 = ?", bound.RoleKey, jobExecutionLogsAPIPath, "GET").Count(&boundPolicy)
|
||||
db.Model(&purchaserCasbinRule{}).Where("v0 = ? AND v1 = ? AND v2 = ?", unbound.RoleKey, jobExecutionLogsAPIPath, "GET").Count(&unboundPolicy)
|
||||
db.Model(&purchaserCasbinRule{}).Where("v0 = ? AND v1 = ?", bound.RoleKey, "/custom").Count(&customPolicy)
|
||||
if apiCount != 1 || menuAPI != 1 || boundPolicy != 1 || unboundPolicy != 0 || customPolicy != 1 {
|
||||
t.Fatalf("unexpected permission state api=%d menu=%d bound=%d unbound=%d custom=%d", apiCount, menuAPI, boundPolicy, unboundPolicy, customPolicy)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEnsureSysJobExecutionLogRequiresExistingMenu(t *testing.T) {
|
||||
db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.AutoMigrate(&migrationmodels.SysMenu{}, &migrationmodels.SysApi{}, &migrationmodels.SysRole{}, &purchaserCasbinRule{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := ensureSysJobExecutionLog(db); err == nil {
|
||||
t.Fatal("missing JobLog menu must fail closed")
|
||||
}
|
||||
}
|
||||
@@ -17,6 +17,15 @@ export function getSysJob(jobId) {
|
||||
})
|
||||
}
|
||||
|
||||
// 查询指定定时任务的持久化执行历史
|
||||
export function listJobExecutionLogs(jobId, query) {
|
||||
return request({
|
||||
url: '/api/v1/sysjob/' + jobId + '/execution-logs',
|
||||
method: 'get',
|
||||
params: query
|
||||
})
|
||||
}
|
||||
|
||||
// 新增SysJob
|
||||
export function addSysJob(data) {
|
||||
return request({
|
||||
@@ -59,4 +68,3 @@ export function startJob(jobId) {
|
||||
method: 'get'
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,24 @@
|
||||
export function jobLogRoute(selectedIds) {
|
||||
if (!Array.isArray(selectedIds) || selectedIds.length !== 1) return null
|
||||
const jobId = Number(selectedIds[0])
|
||||
if (!Number.isInteger(jobId) || jobId < 1) return null
|
||||
return { name: 'JobLog', query: { jobId: String(jobId) }}
|
||||
}
|
||||
|
||||
export function executionStatusMeta(status) {
|
||||
return {
|
||||
running: { label: '执行中', type: 'primary' },
|
||||
succeeded: { label: '成功', type: 'success' },
|
||||
failed: { label: '失败', type: 'danger' },
|
||||
interrupted: { label: '已中断', type: 'warning' }
|
||||
}[status] || { label: status || '未知', type: 'info' }
|
||||
}
|
||||
|
||||
export function executionLogQuery(query, dateRange) {
|
||||
const result = { ...query }
|
||||
if (Array.isArray(dateRange) && dateRange.length === 2) {
|
||||
result.startedFrom = new Date(dateRange[0]).toISOString()
|
||||
result.startedTo = new Date(dateRange[1]).toISOString()
|
||||
}
|
||||
return result
|
||||
}
|
||||
@@ -55,7 +55,7 @@
|
||||
<el-button v-permisaction="['job:sysJob:add']" type="primary" size="small" :icon="Plus" @click="handleAdd">新增</el-button>
|
||||
<el-button v-permisaction="['job:sysJob:edit']" type="primary" size="small" :icon="Edit" :disabled="single" @click="handleUpdate">修改</el-button>
|
||||
<el-button v-permisaction="['job:sysJob:remove']" type="danger" size="small" :icon="Delete" :disabled="multiple" @click="handleDelete">删除</el-button>
|
||||
<el-button v-permisaction="['job:sysJob:log']" size="small" @click="handleLog">日志</el-button>
|
||||
<el-button v-permisaction="['job:sysJob:log']" size="small" :disabled="single" @click="handleLog">日志</el-button>
|
||||
</div>
|
||||
|
||||
<el-table v-loading="loading" :data="sysjobList" border stripe @selection-change="handleSelectionChange">
|
||||
@@ -267,6 +267,7 @@
|
||||
<script>
|
||||
import { addSysJob, delSysJob, getSysJob, listSysJob, updateSysJob, removeJob, startJob } from '@/api/job/sys-job'
|
||||
import { Search, Refresh, Plus, Edit, Delete } from '@element-plus/icons-vue'
|
||||
import { jobLogRoute } from './execution-log'
|
||||
|
||||
export default {
|
||||
name: 'SysJobManage',
|
||||
@@ -500,7 +501,8 @@ export default {
|
||||
}).catch(function() {})
|
||||
},
|
||||
handleLog() {
|
||||
this.$router.push({ name: 'job_log', params: { }})
|
||||
const route = jobLogRoute(this.ids)
|
||||
if (route) this.$router.push(route)
|
||||
},
|
||||
handleSybSyncRuns() {
|
||||
this.$router.push('/syb-sync-runs/index')
|
||||
|
||||
@@ -1,101 +1,116 @@
|
||||
|
||||
<template>
|
||||
<BasicLayout>
|
||||
<template #wrapper>
|
||||
<el-card class="box-card">
|
||||
<el-form>
|
||||
<el-card class="page-card" shadow="never">
|
||||
<div class="page-heading">
|
||||
<div>
|
||||
<div class="eyebrow">定时任务 / 执行日志</div>
|
||||
<h1>{{ job.jobName || '定时任务执行日志' }}<span v-if="job.jobId">（#{{ job.jobId }}）</span></h1>
|
||||
<p>查看该任务每次调度的开始时间、结果、耗时和脱敏错误摘要。</p>
|
||||
</div>
|
||||
<el-button :icon="ArrowLeft" @click="backToList">返回任务列表</el-button>
|
||||
</div>
|
||||
|
||||
<el-alert v-if="job.deleted" title="该定时任务已删除，以下为保留的历史执行记录。" type="warning" show-icon :closable="false" class="notice" role="status" />
|
||||
<el-form :model="query" :inline="true" class="search-form" @submit.prevent="search">
|
||||
<el-form-item label="状态">
|
||||
<el-select v-model="query.status" clearable placeholder="全部状态" style="width:150px">
|
||||
<el-option v-for="item in statusOptions" :key="item.value" :label="item.label" :value="item.value" />
|
||||
</el-select>
|
||||
</el-form-item>
|
||||
<el-form-item label="开始时间">
|
||||
<el-date-picker v-model="dateRange" type="datetimerange" range-separator="至" start-placeholder="开始时间" end-placeholder="结束时间" />
|
||||
</el-form-item>
|
||||
<el-form-item>
|
||||
<el-button type="success" size="mini">状态</el-button>
|
||||
<el-button type="primary" size="mini">清空</el-button>
|
||||
<el-button type="primary" :icon="Search" :disabled="!jobId" @click="search">查询</el-button>
|
||||
<el-button :icon="RefreshLeft" :disabled="!jobId" @click="reset">重置</el-button>
|
||||
</el-form-item>
|
||||
</el-form>
|
||||
<el-row ref="log" :gutter="10" class="mb8">
|
||||
<el-scrollbar style="height:500px;background-color: black;color: cornflowerblue;">
|
||||
<ul
|
||||
style="line-height: 25px;padding-top: 15px;padding-bottom: 15px;min-height: 500px; margin: 0;list-style-type: none;"
|
||||
>
|
||||
<li v-for="(item,index) in arrs" :key="index">
|
||||
|
||||
{{ item }}
|
||||
</li>
|
||||
</ul>
|
||||
</el-scrollbar>
|
||||
</el-row>
|
||||
<el-alert v-if="loadError" :title="loadError" type="error" show-icon :closable="false" class="notice" role="alert">
|
||||
<template #default><el-button v-if="jobId" link type="primary" @click="load">重新加载</el-button></template>
|
||||
</el-alert>
|
||||
<el-table v-loading="loading" :data="items" border stripe :empty-text="jobId ? '该任务尚无执行记录' : '请从定时任务列表选择一条任务'">
|
||||
<el-table-column label="执行标识" prop="executionId" min-width="205" show-overflow-tooltip />
|
||||
<el-table-column label="开始时间" min-width="175"><template #default="{ row }">{{ formatTime(row.startedAt) }}</template></el-table-column>
|
||||
<el-table-column label="结束时间" min-width="175"><template #default="{ row }">{{ row.status === 'running' ? '执行中' : formatTime(row.finishedAt) }}</template></el-table-column>
|
||||
<el-table-column label="状态" width="105"><template #default="{ row }"><el-tag :type="statusMeta(row.status).type">{{ statusMeta(row.status).label }}</el-tag></template></el-table-column>
|
||||
<el-table-column label="耗时" width="105"><template #default="{ row }">{{ row.status === 'running' ? '执行中' : formatDuration(row.durationMs) }}</template></el-table-column>
|
||||
<el-table-column label="触发方式" width="110"><template #default="{ row }">{{ row.triggerType === 'scheduled' ? '定时调度' : row.triggerType }}</template></el-table-column>
|
||||
<el-table-column label="错误摘要" min-width="260"><template #default="{ row }"><span v-if="row.errorCode" class="error-code">{{ row.errorCode }}</span>{{ row.errorMessage || '—' }}</template></el-table-column>
|
||||
</el-table>
|
||||
<pagination v-show="total > 0" v-model:current-page="query.pageIndex" v-model:page-size="query.pageSize" :total="total" @pagination="load" />
|
||||
</el-card>
|
||||
</template>
|
||||
</BasicLayout>
|
||||
|
||||
</template>
|
||||
|
||||
<script>
|
||||
import { ArrowLeft, RefreshLeft, Search } from '@element-plus/icons-vue'
|
||||
import { listJobExecutionLogs } from '@/api/job/sys-job'
|
||||
import { executionLogQuery, executionStatusMeta } from './execution-log'
|
||||
|
||||
import { unWsLogout } from '@/api/ws'
|
||||
export default {
|
||||
name: 'SysJobLogManage',
|
||||
setup() { return { ArrowLeft, RefreshLeft, Search } },
|
||||
data() {
|
||||
return {
|
||||
websock: null,
|
||||
arrs: [],
|
||||
id: undefined,
|
||||
group: undefined
|
||||
loading: false, loadError: '', job: {}, items: [], total: 0, dateRange: null, pollTimer: null,
|
||||
query: { pageIndex: 1, pageSize: 20, status: '' },
|
||||
statusOptions: [
|
||||
{ label: '执行中', value: 'running' }, { label: '成功', value: 'succeeded' },
|
||||
{ label: '失败', value: 'failed' }, { label: '已中断', value: 'interrupted' }
|
||||
]
|
||||
}
|
||||
},
|
||||
computed: {
|
||||
jobId() {
|
||||
const value = Number(this.$route.query.jobId)
|
||||
return Number.isInteger(value) && value > 0 ? value : 0
|
||||
}
|
||||
},
|
||||
created() {
|
||||
this.id = this.guid()
|
||||
this.group = 'log'
|
||||
this.initWebSocket()
|
||||
},
|
||||
unmounted() {
|
||||
console.log('断开websocket连接')
|
||||
this.websock.close() // 离开路由之后断开websocket连接
|
||||
unWsLogout(this.id, this.group).then(response => {
|
||||
console.log(response.data)
|
||||
if (!this.jobId) {
|
||||
this.loadError = '缺少有效的定时任务编号，请返回列表重新选择。'
|
||||
return
|
||||
}
|
||||
)
|
||||
this.load()
|
||||
},
|
||||
beforeUnmount() { this.stopPolling() },
|
||||
methods: {
|
||||
initWebSocket() { // 初始化weosocket
|
||||
console.log(this.$store.state.user.token)
|
||||
const wsuri = 'ws://127.0.0.1:8000/ws/' + this.id + '/' + this.group + '?token=' + this.$store.state.user.token
|
||||
this.websock = new WebSocket(wsuri)
|
||||
this.websock.onmessage = this.websocketonmessage
|
||||
this.websock.onopen = this.websocketonopen
|
||||
this.websock.onerror = this.websocketonerror
|
||||
this.websock.onclose = this.websocketclose
|
||||
statusMeta: executionStatusMeta,
|
||||
formatTime(value) { return value ? new Date(value).toLocaleString('zh-CN', { hour12: false }) : '—' },
|
||||
formatDuration(value) {
|
||||
const ms = Number(value) || 0
|
||||
if (ms < 1000) return `${ms}ms`
|
||||
if (ms < 60000) return `${(ms / 1000).toFixed(1)}s`
|
||||
return `${Math.floor(ms / 60000)}m ${Math.round((ms % 60000) / 1000)}s`
|
||||
},
|
||||
websocketonopen() { // 连接建立之后执行send方法发送数据
|
||||
console.log('连接打开')
|
||||
// const actions = { 'test': '12345' }
|
||||
// this.websocketsend(JSON.stringify(actions))
|
||||
},
|
||||
websocketonerror() { // 连接建立失败重连
|
||||
this.initWebSocket()
|
||||
},
|
||||
websocketonmessage(e) { // 数据接收
|
||||
console.log(e.data)
|
||||
// console.log(this.binaryAgent(e))
|
||||
// const redata = JSON.parse(e.data)
|
||||
// console.log(redata)
|
||||
// this.$refs.log.innerText = e.data + '\n' + this.$refs.log.innerText
|
||||
this.arrs.unshift(e.data)
|
||||
},
|
||||
websocketsend(Data) { // 数据发送
|
||||
// this.websock.send(Data)
|
||||
},
|
||||
websocketclose(e) { // 关闭
|
||||
unWsLogout(this.id, this.group).then(response => {
|
||||
console.log(response.data)
|
||||
async load() {
|
||||
if (!this.jobId) return
|
||||
this.loading = true
|
||||
this.loadError = ''
|
||||
try {
|
||||
const response = await listJobExecutionLogs(this.jobId, executionLogQuery(this.query, this.dateRange))
|
||||
this.job = response.data.job
|
||||
this.items = response.data.items
|
||||
this.total = response.data.total
|
||||
this.items.some(item => item.status === 'running') ? this.startPolling() : this.stopPolling()
|
||||
} catch (error) {
|
||||
this.loadError = error?.response?.data?.msg || error?.message || '执行日志加载失败'
|
||||
} finally {
|
||||
this.loading = false
|
||||
}
|
||||
)
|
||||
console.log('断开连接', e)
|
||||
},
|
||||
guid() {
|
||||
return 'xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx'.replace(/[xy]/g, function(c) {
|
||||
var r = Math.random() * 16 | 0; var v = c === 'x' ? r : (r & 0x3 | 0x8)
|
||||
return v.toString(16)
|
||||
})
|
||||
}
|
||||
search() { this.query.pageIndex = 1; this.load() },
|
||||
reset() { this.dateRange = null; this.query = { pageIndex: 1, pageSize: 20, status: '' }; this.load() },
|
||||
backToList() { this.$router.push('/schedule/manage') },
|
||||
startPolling() { if (!this.pollTimer) this.pollTimer = window.setInterval(() => this.load(), 5000) },
|
||||
stopPolling() { if (this.pollTimer) { window.clearInterval(this.pollTimer); this.pollTimer = null } }
|
||||
}
|
||||
}
|
||||
|
||||
</script>
|
||||
|
||||
<style lang="scss" scoped>
|
||||
.page-card{min-height:calc(100vh - 124px)}.page-heading{display:flex;align-items:flex-start;justify-content:space-between;gap:16px;margin-bottom:16px}.page-heading h1{margin:4px 0 6px;color:var(--el-text-color-primary);font-size:24px}.page-heading p{margin:0;color:var(--el-text-color-regular);line-height:1.5}.eyebrow{color:var(--el-text-color-secondary);font-size:13px}.search-form{padding:16px 16px 0;margin-bottom:16px;border:1px solid var(--el-border-color);border-radius:8px;background:var(--el-fill-color-lighter)}.notice{margin-bottom:16px}.error-code{display:inline-block;margin-right:8px;color:var(--el-color-danger);font-family:Consolas,monospace;font-size:12px}@media(max-width:800px){.page-heading{flex-direction:column}.search-form :deep(.el-date-editor){width:100%}}
|
||||
</style>
|
||||
|
||||
@@ -0,0 +1,39 @@
|
||||
import { expect, test } from '@playwright/test'
|
||||
|
||||
async function authenticate(context: any) {
|
||||
await context.addCookies([{ name: 'Admin-Token', value: 'job-log-test-token', domain: 'localhost', path: '/' }])
|
||||
}
|
||||
|
||||
test('单选定时任务后进入对应的持久化执行日志', async ({ page, context }) => {
|
||||
await authenticate(context)
|
||||
await page.route('**/api/**', async route => {
|
||||
const url = new URL(route.request().url())
|
||||
if (url.pathname.startsWith('/src/api/')) return route.continue()
|
||||
if (url.pathname.endsWith('/api/v1/getinfo')) {
|
||||
return route.fulfill({ json: { code: 200, data: { roles: ['admin'], name: '管理员', avatar: '', introduction: '', permissions: ['job:sysJob:log'] } } })
|
||||
}
|
||||
if (url.pathname.endsWith('/api/v1/menurole')) {
|
||||
return route.fulfill({ json: { code: 200, data: [{ path: '/schedule', component: 'Layout', visible: '0', menuName: 'Schedule', title: '定时任务', icon: 'time', children: [{ path: 'manage', component: '/schedule/index', visible: '0', menuName: 'ScheduleManage', title: '定时任务' }, { path: 'log', component: '/schedule/log', visible: '1', menuName: 'JobLog', title: '执行日志' }] }] } })
|
||||
}
|
||||
if (url.pathname.endsWith('/api/v1/sysjob/5/execution-logs')) {
|
||||
return route.fulfill({ json: { code: 200, data: { job: { jobId: 5, jobName: 'SYB 规格 AI 修复', invokeTarget: 'GoAutoSYBSpecAIParse', deleted: false }, items: [{ id: 21, executionId: '00000000-0000-4000-8000-000000000021', jobId: 5, jobName: 'SYB 规格 AI 修复', invokeTarget: 'GoAutoSYBSpecAIParse', triggerType: 'scheduled', status: 'failed', startedAt: '2026-09-02T01:00:00Z', finishedAt: '2026-09-02T01:00:03Z', durationMs: 3000, errorCode: 'JOB_EXECUTION_FAILED', errorMessage: '任务执行失败，请查看受控服务日志' }], total: 1, page: 1, pageSize: 20 } } })
|
||||
}
|
||||
if (url.pathname.endsWith('/api/v1/sysjob')) {
|
||||
return route.fulfill({ json: { code: 200, data: { list: [{ jobId: 5, jobName: 'SYB 规格 AI 修复', jobGroup: 'GoAuto', cronExpression: '0 0 * * * *', invokeTarget: 'GoAutoSYBSpecAIParse', status: 2, entry_id: 0 }], count: 1 } } })
|
||||
}
|
||||
return route.fulfill({ json: { code: 200, data: [] } })
|
||||
})
|
||||
|
||||
await page.goto('/#/schedule/manage')
|
||||
const logButton = page.getByRole('button', { name: '日志', exact: true })
|
||||
await expect(page.getByText('SYB 规格 AI 修复', { exact: true })).toBeVisible()
|
||||
await expect(logButton).toBeDisabled()
|
||||
await page.locator('.el-table__body-wrapper .el-checkbox').first().click()
|
||||
await expect(logButton).toBeEnabled()
|
||||
await logButton.click()
|
||||
|
||||
await expect(page).toHaveURL(/#\/schedule\/log\?jobId=5$/)
|
||||
await expect(page.getByRole('heading', { name: 'SYB 规格 AI 修复（#5）' })).toBeVisible()
|
||||
await expect(page.getByText('JOB_EXECUTION_FAILED')).toBeVisible()
|
||||
await expect(page.getByText('任务执行失败，请查看受控服务日志')).toBeVisible()
|
||||
})
|
||||
@@ -0,0 +1,26 @@
|
||||
import { executionLogQuery, executionStatusMeta, jobLogRoute } from '@/views/schedule/execution-log'
|
||||
|
||||
describe('scheduled job execution log helpers', () => {
|
||||
test('only one valid selection can navigate to JobLog', () => {
|
||||
expect(jobLogRoute([])).toBeNull()
|
||||
expect(jobLogRoute([1, 2])).toBeNull()
|
||||
expect(jobLogRoute(['invalid'])).toBeNull()
|
||||
expect(jobLogRoute([5])).toEqual({ name: 'JobLog', query: { jobId: '5' }})
|
||||
})
|
||||
|
||||
test('builds RFC3339 date filters without mutating paging', () => {
|
||||
const from = new Date('2026-09-02T10:00:00+08:00')
|
||||
const to = new Date('2026-09-02T11:00:00+08:00')
|
||||
expect(executionLogQuery({ pageIndex: 2, pageSize: 20, status: 'failed' }, [from, to])).toEqual({
|
||||
pageIndex: 2, pageSize: 20, status: 'failed',
|
||||
startedFrom: from.toISOString(), startedTo: to.toISOString()
|
||||
})
|
||||
})
|
||||
|
||||
test('maps all persisted statuses', () => {
|
||||
expect(executionStatusMeta('running').label).toBe('执行中')
|
||||
expect(executionStatusMeta('succeeded').type).toBe('success')
|
||||
expect(executionStatusMeta('failed').type).toBe('danger')
|
||||
expect(executionStatusMeta('interrupted').label).toBe('已中断')
|
||||
})
|
||||
})
|
||||
Reference in New Issue
Block a user