opensource/certimate
3
0
Fork 0
mirror of https://github.com/certimate-go/certimate.git synced 2026-09-01 15:39:35 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: optimize workflow dispatcher

Browse Source
This commit is contained in:
Fu Diwei
2025-12-11 11:21:20 +08:00
committed by RHQYZ
parent ae06314cf5
commit 456b2cda5c
6 changed files with 68 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/workflow/dispatcher/dispatcher.go
+13 -16
View File
@@ -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) }()
internal/workflow/engine/executor_bizapply.go
+1 -1
View File
@@ -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 {
internal/workflow/engine/executor_bizdeploy.go
+1 -1
View File
@@ -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))
}
}
internal/workflow/event.go → internal/workflow/pbhook.go
+5 -24
View File
@@ -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
internal/workflow/pbjob.go
+43
View File
@@ -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
}
internal/workflow/service.go
+5 -19
View File
@@ -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