From 456b2cda5c9d3920b9d8d580563ca25b356ec5c8 Mon Sep 17 00:00:00 2001 From: Fu Diwei Date: Thu, 11 Dec 2025 11:21:20 +0800 Subject: [PATCH] feat: optimize workflow dispatcher --- internal/workflow/dispatcher/dispatcher.go | 29 ++++++------- internal/workflow/engine/executor_bizapply.go | 2 +- .../workflow/engine/executor_bizdeploy.go | 2 +- internal/workflow/{event.go => pbhook.go} | 29 +++---------- internal/workflow/pbjob.go | 43 +++++++++++++++++++ internal/workflow/service.go | 24 +++-------- 6 files changed, 68 insertions(+), 61 deletions(-) rename internal/workflow/{event.go => pbhook.go} (61%) create mode 100644 internal/workflow/pbjob.go diff --git a/internal/workflow/dispatcher/dispatcher.go b/internal/workflow/dispatcher/dispatcher.go index 9f8ff769b..253e1d842 100644 --- a/internal/workflow/dispatcher/dispatcher.go +++ b/internal/workflow/dispatcher/dispatcher.go @@ -127,23 +127,20 @@ func (wd *workflowDispatcher) Shutdown(ctx context.Context) error { } func (wd *workflowDispatcher) Start(ctx context.Context, runId string) error { - wd.taskMtx.RLock() + wd.taskMtx.Lock() + defer wd.taskMtx.Unlock() + if _, exists := wd.processingTasks[runId]; exists { - wd.taskMtx.RUnlock() return fmt.Errorf("workflow run %s is already processing", runId) } + for _, pendingRunId := range wd.pendingRunQueue { if pendingRunId == runId { - wd.taskMtx.RUnlock() return fmt.Errorf("workflow run %s is already in the queue", runId) } } - wd.taskMtx.RUnlock() - wd.taskMtx.Lock() wd.pendingRunQueue = append(wd.pendingRunQueue, runId) - wd.taskMtx.Unlock() - go func() { wd.tryNextAsync() }() return nil @@ -157,7 +154,7 @@ func (wd *workflowDispatcher) Cancel(ctx context.Context, runId string) error { if err != nil { return err } else if workflowRun.Status != domain.WorkflowRunStatusTypePending && workflowRun.Status != domain.WorkflowRunStatusTypeProcessing { - return fmt.Errorf("workflow run #%s is already completed", workflowRun.Id) + return fmt.Errorf("workrun #%s is already completed", workflowRun.Id) } workflow, err := wd.workflowRepo.GetById(ctx, workflowRun.WorkflowId) @@ -182,7 +179,7 @@ func (wd *workflowDispatcher) Cancel(ctx context.Context, runId string) error { task.cancel() delete(wd.processingTasks, runId) - wd.syslog.Info(fmt.Sprintf("workflow run #%s was canceled", task.RunId)) + wd.syslog.Info(fmt.Sprintf("workrun #%s was canceled", task.RunId)) } for i, pendingRunId := range wd.pendingRunQueue { @@ -231,7 +228,7 @@ func (wd *workflowDispatcher) tryExecuteAsync(task *taskInfo) { // 查询运行实体，并级联更新状态 if workflowRun, err = wd.workflowRunRepo.GetById(task.ctx, task.RunId); err != nil { if !(errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)) { - wd.syslog.Error(fmt.Sprintf("failed to get workflow run #%s record", task.RunId), slog.Any("error", err)) + wd.syslog.Error(fmt.Sprintf("failed to get workrun #%s record", task.RunId), slog.Any("error", err)) } return } else { @@ -321,7 +318,7 @@ func (wd *workflowDispatcher) tryExecuteAsync(task *taskInfo) { }) // 执行工作流 - wd.syslog.Info(fmt.Sprintf("workflow run #%s (work#%s) started", task.RunId, task.WorkflowId)) + wd.syslog.Info(fmt.Sprintf("workflow #%s's run #%s started", task.WorkflowId, task.RunId)) we.Invoke(task.ctx, engine.WorkflowExecution{ WorkflowId: workflowRun.WorkflowId, WorkflowName: workflow.Name, @@ -329,7 +326,7 @@ func (wd *workflowDispatcher) tryExecuteAsync(task *taskInfo) { RunTrigger: workflowRun.Trigger, Graph: workflowRun.Graph, }) - wd.syslog.Info(fmt.Sprintf("workflow run #%s (work#%s) stopped", task.RunId, task.WorkflowId)) + wd.syslog.Info(fmt.Sprintf("workflow #%s's run #%s stopped", task.WorkflowId, task.RunId)) } func (wd *workflowDispatcher) tryNextAsync() { @@ -338,7 +335,7 @@ func (wd *workflowDispatcher) tryNextAsync() { for _, pendingRunId := range wd.pendingRunQueue { workflowRun, err := wd.workflowRunRepo.GetById(context.Background(), pendingRunId) if err != nil { - wd.syslog.Error(fmt.Sprintf("failed to get workflow run #%s record", pendingRunId), slog.Any("error", err)) + wd.syslog.Error(fmt.Sprintf("failed to get workrun #%s record", pendingRunId), slog.Any("error", err)) continue } @@ -351,9 +348,9 @@ func (wd *workflowDispatcher) tryNextAsync() { } if hasSameWorkflowTask { - wd.syslog.Warn(fmt.Sprintf("workflow run #%s is pending, because tasks that belonging to the same workflow #%s already exists", workflowRun.Id, workflowRun.WorkflowId)) + wd.syslog.Warn(fmt.Sprintf("workflow #%s's run #%s is pending, because tasks that belonging to the same workflow already exists", workflowRun.WorkflowId, workflowRun.Id)) } else if len(wd.processingTasks) >= wd.concurrency && wd.concurrency > 0 { - wd.syslog.Warn(fmt.Sprintf("workflow run #%s is pending, because the maximum concurrency (limit: %d) has been reached", pendingRunId, wd.concurrency)) + wd.syslog.Warn(fmt.Sprintf("workflow #%s's run #%s is pending, because the maximum concurrency (limit: %d) has been reached", workflowRun.WorkflowId, workflowRun.Id, wd.concurrency)) } else { wd.taskMtx.RUnlock() @@ -362,7 +359,7 @@ func (wd *workflowDispatcher) tryNextAsync() { task := &taskInfo{WorkflowId: workflowRun.WorkflowId, RunId: workflowRun.Id, ctx: ctxRun, cancel: ctxCancel} wd.pendingRunQueue = lo.Filter(wd.pendingRunQueue, func(s string, _ int) bool { return s != pendingRunId }) wd.processingTasks[pendingRunId] = task - wd.syslog.Info(fmt.Sprintf("workflow run #%s (work#%s) is being dispatched ...", task.RunId, task.WorkflowId)) + wd.syslog.Info(fmt.Sprintf("workflow #%s's run #%s is being dispatched ...", task.WorkflowId, task.RunId)) wd.taskMtx.Unlock() go func() { wd.tryExecuteAsync(task) }() diff --git a/internal/workflow/engine/executor_bizapply.go b/internal/workflow/engine/executor_bizapply.go index 248520027..581a822f8 100644 --- a/internal/workflow/engine/executor_bizapply.go +++ b/internal/workflow/engine/executor_bizapply.go @@ -75,7 +75,7 @@ func (ne *bizApplyNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeExe return execRes, err } else { if lastOutput != nil { - ne.logger.Info(fmt.Sprintf("found last workflow run #%s", lastOutput.RunId)) + ne.logger.Info(fmt.Sprintf("found last run #%s", lastOutput.RunId)) } if lastCertificate != nil { diff --git a/internal/workflow/engine/executor_bizdeploy.go b/internal/workflow/engine/executor_bizdeploy.go index a6561e4dc..e6d892578 100644 --- a/internal/workflow/engine/executor_bizdeploy.go +++ b/internal/workflow/engine/executor_bizdeploy.go @@ -38,7 +38,7 @@ func (ne *bizDeployNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeEx return execRes, err } else { if lastOutput != nil { - ne.logger.Info(fmt.Sprintf("found last workflow run #%s", lastOutput.RunId)) + ne.logger.Info(fmt.Sprintf("found last run #%s", lastOutput.RunId)) } } diff --git a/internal/workflow/event.go b/internal/workflow/pbhook.go similarity index 61% rename from internal/workflow/event.go rename to internal/workflow/pbhook.go index 81e1db22d..c9029dc20 100644 --- a/internal/workflow/event.go +++ b/internal/workflow/pbhook.go @@ -3,15 +3,11 @@ package workflow import ( "context" "fmt" - "log/slog" "github.com/pocketbase/pocketbase/core" - "github.com/pocketbase/pocketbase/tools/cron" - "github.com/samber/lo" "github.com/certimate-go/certimate/internal/app" "github.com/certimate-go/certimate/internal/domain" - "github.com/certimate-go/certimate/internal/domain/dtos" "github.com/certimate-go/certimate/internal/repository" ) @@ -55,39 +51,24 @@ func Register() { }) } -func onWorkflowRecordCreateOrUpdate(ctx context.Context, record *core.Record) error { +func onWorkflowRecordCreateOrUpdate(_ context.Context, record *core.Record) error { scheduler := app.GetScheduler() // 向数据库插入/更新时，同时更新定时任务 - jobId := fmt.Sprintf("workflow#%s", record.Id) enabled := record.GetBool("enabled") trigger := record.GetString("trigger") triggerCron := record.GetString("triggerCron") // 如果非定时触发或未启用，移除定时任务 if !enabled || trigger != string(domain.WorkflowTriggerTypeScheduled) { - scheduler.Remove(jobId) + scheduler.Remove(fmt.Sprintf("workflow#%s", record.Id)) return nil } // 反之，重新添加定时任务 - job, _ := lo.Find(scheduler.Jobs(), func(j *cron.Job) bool { return j.Id() == jobId }) - if job == nil || job.Expression() != triggerCron { - workflowId := record.Id - err := scheduler.Add(jobId, triggerCron, func() { - workflowSrv := NewWorkflowService(repository.NewWorkflowRepository(), repository.NewWorkflowRunRepository(), repository.NewSettingsRepository()) - _, err := workflowSrv.StartRun(context.Background(), &dtos.WorkflowStartRunReq{ - WorkflowId: workflowId, - RunTrigger: domain.WorkflowTriggerTypeScheduled, - }) - if err != nil { - app.GetLogger().Warn(fmt.Sprintf("failed to start scheduled run for workflow #%s", workflowId), slog.Any("error", err)) - } - }) - if err != nil { - app.GetLogger().Error(fmt.Sprintf("failed to register cron job for workflow #%s", workflowId), slog.Any("error", err)) - return fmt.Errorf("failed to add cron job: %w", err) - } + workflowSrv := NewWorkflowService(repository.NewWorkflowRepository(), repository.NewWorkflowRunRepository(), repository.NewSettingsRepository()) + if err := addWorkflowJob(workflowSrv, record.Id, triggerCron); err != nil { + return err } return nil diff --git a/internal/workflow/pbjob.go b/internal/workflow/pbjob.go new file mode 100644 index 000000000..11786bb23 --- /dev/null +++ b/internal/workflow/pbjob.go @@ -0,0 +1,43 @@ +package workflow + +import ( + "context" + "fmt" + "log/slog" + + "github.com/pocketbase/pocketbase/tools/cron" + "github.com/samber/lo" + + "github.com/certimate-go/certimate/internal/app" + "github.com/certimate-go/certimate/internal/domain" + "github.com/certimate-go/certimate/internal/domain/dtos" +) + +func addWorkflowJob(workflowSrv *WorkflowService, workflowId string, triggerCron string) error { + scheduler := app.GetScheduler() + + jobId := fmt.Sprintf("workflow#%s", workflowId) + job, _ := lo.Find(scheduler.Jobs(), func(j *cron.Job) bool { return j.Id() == jobId }) + if job != nil && job.Expression() == triggerCron { + return nil + } + + err := scheduler.Add(jobId, triggerCron, func() { + app.GetLogger().Info(fmt.Sprintf("workflow #%s is triggered ...", workflowId)) + + _, err := workflowSrv.StartRun(context.Background(), &dtos.WorkflowStartRunReq{ + WorkflowId: workflowId, + RunTrigger: domain.WorkflowTriggerTypeScheduled, + }) + if err != nil { + app.GetLogger().Warn(fmt.Sprintf("failed to start scheduled run for workflow #%s", workflowId), slog.Any("error", err)) + } + }) + if err != nil { + app.GetLogger().Error(fmt.Sprintf("failed to register cron job for workflow #%s", workflowId), slog.Any("error", err)) + return fmt.Errorf("failed to add cron job: %w", err) + } + + app.GetLogger().Info(fmt.Sprintf("registered cron job for workflow #%s", workflowId), slog.String("cron", triggerCron)) + return nil +} diff --git a/internal/workflow/service.go b/internal/workflow/service.go index 3fcb75e64..88e708a02 100644 --- a/internal/workflow/service.go +++ b/internal/workflow/service.go @@ -52,29 +52,15 @@ func (s *WorkflowService) InitSchedule(ctx context.Context) error { return err } + var errs []error for _, workflow := range workflows { - var errs []error - - err := app.GetScheduler().Add(fmt.Sprintf("workflow#%s", workflow.Id), workflow.TriggerCron, func() { - _, err := s.StartRun(context.Background(), &dtos.WorkflowStartRunReq{ - WorkflowId: workflow.Id, - RunTrigger: domain.WorkflowTriggerTypeScheduled, - }) - if err != nil { - app.GetLogger().Error(fmt.Sprintf("failed to start scheduled run for workflow #%s", workflow.Id), slog.Any("error", err)) - } - }) - if err != nil { - app.GetLogger().Error(fmt.Sprintf("failed to register cron job for workflow #%s", workflow.Id), slog.Any("error", err)) + if err := addWorkflowJob(s, workflow.Id, workflow.TriggerCron); err != nil { errs = append(errs, err) - } else { - app.GetLogger().Info(fmt.Sprintf("registered cron job for workflow #%s", workflow.Id), slog.String("cron", workflow.TriggerCron)) - } - - if len(errs) > 0 { - return errors.Join(errs...) } } + if len(errs) > 0 { + return errors.Join(errs...) + } } return nil