opensource/certimate
3
0
Fork 0
mirror of https://github.com/certimate-go/certimate.git synced 2026-09-21 12:43:46 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor: make workflow node config persistence in outputs

Browse Source
This commit is contained in:
Fu Diwei
2025-08-21 10:24:56 +08:00
parent 4fb09b3ab1
commit 1bc92b5b78
15 changed files with 149 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/applicant/applicant.go
+1 -1
View File
@@ -54,7 +54,7 @@ func NewWithWorkflowNode(config ApplicantWithWorkflowNodeConfig) (Applicant, err
return nil, fmt.Errorf("the node type is '%s', expected '%s'", string(config.Node.Type), string(domain.WorkflowNodeTypeBizApply))
}
nodeCfg := config.Node.GetConfigForBizApply()
nodeCfg := config.Node.Data.Config.AsBizApply()
options := &applicantProviderOptions{
Domains: lo.Filter(strings.Split(nodeCfg.Domains, ";"), func(s string, _ int) bool { return s != "" }),
ContactEmail: nodeCfg.ContactEmail,
internal/deployer/deployer.go
+1 -1
View File
@@ -29,7 +29,7 @@ func NewWithWorkflowNode(config DeployerWithWorkflowNodeConfig) (Deployer, error
return nil, fmt.Errorf("the node type is '%s', expected '%s'", string(config.Node.Type), string(domain.WorkflowNodeTypeBizDeploy))
}
nodeCfg := config.Node.GetConfigForBizDeploy()
nodeCfg := config.Node.Data.Config.AsBizDeploy()
options := &deployerProviderOptions{
Provider: domain.DeploymentProviderType(nodeCfg.Provider),
ProviderAccessConfig: make(map[string]any),
internal/domain/workflow.go
+83 -81
View File
@@ -95,8 +95,89 @@ const (
)
type WorkflowNodeData struct {
Name string `json:"name"`
Config map[string]any `json:"config"`
Name string `json:"name"`
Config WorkflowNodeConfig `json:"config"`
}
type WorkflowNodeConfig map[string]any
func (c WorkflowNodeConfig) AsBizApply() WorkflowNodeConfigForBizApply {
return WorkflowNodeConfigForBizApply{
Domains: xmaps.GetString(c, "domains"),
ContactEmail: xmaps.GetString(c, "contactEmail"),
ChallengeType: xmaps.GetString(c, "challengeType"),
Provider: xmaps.GetString(c, "provider"),
ProviderAccessId: xmaps.GetString(c, "providerAccessId"),
ProviderConfig: xmaps.GetKVMapAny(c, "providerConfig"),
KeyAlgorithm: xmaps.GetOrDefaultString(c, "keyAlgorithm", string(CertificateKeyAlgorithmTypeRSA2048)),
CAProvider: xmaps.GetString(c, "caProvider"),
CAProviderAccessId: xmaps.GetString(c, "caProviderAccessId"),
CAProviderConfig: xmaps.GetKVMapAny(c, "caProviderConfig"),
ACMEProfile: xmaps.GetString(c, "acmeProfile"),
Nameservers: xmaps.GetString(c, "nameservers"),
DnsPropagationWait: xmaps.GetInt32(c, "dnsPropagationWait"),
DnsPropagationTimeout: xmaps.GetInt32(c, "dnsPropagationTimeout"),
DnsTTL: xmaps.GetInt32(c, "dnsTTL"),
DisableFollowCNAME: xmaps.GetBool(c, "disableFollowCNAME"),
DisableARI: xmaps.GetBool(c, "disableARI"),
SkipBeforeExpiryDays: xmaps.GetInt32(c, "skipBeforeExpiryDays"),
}
}
func (c WorkflowNodeConfig) AsBizUpload() WorkflowNodeConfigForBizUpload {
return WorkflowNodeConfigForBizUpload{
Certificate: xmaps.GetString(c, "certificate"),
PrivateKey: xmaps.GetString(c, "privateKey"),
Domains: xmaps.GetString(c, "domains"),
}
}
func (c WorkflowNodeConfig) AsBizMonitor() WorkflowNodeConfigForBizMonitor {
host := xmaps.GetString(c, "host")
return WorkflowNodeConfigForBizMonitor{
Host: host,
Port: xmaps.GetOrDefaultInt32(c, "port", 443),
Domain: xmaps.GetOrDefaultString(c, "domain", host),
RequestPath: xmaps.GetString(c, "path"),
}
}
func (c WorkflowNodeConfig) AsBizDeploy() WorkflowNodeConfigForBizDeploy {
return WorkflowNodeConfigForBizDeploy{
CertificateOutputNodeId: xmaps.GetString(c, "certificateOutputNodeId"),
Provider: xmaps.GetString(c, "provider"),
ProviderAccessId: xmaps.GetString(c, "providerAccessId"),
ProviderConfig: xmaps.GetKVMapAny(c, "providerConfig"),
SkipOnLastSucceeded: xmaps.GetBool(c, "skipOnLastSucceeded"),
}
}
func (c WorkflowNodeConfig) AsBizNotify() WorkflowNodeConfigForBizNotify {
return WorkflowNodeConfigForBizNotify{
Provider: xmaps.GetString(c, "provider"),
ProviderAccessId: xmaps.GetString(c, "providerAccessId"),
ProviderConfig: xmaps.GetKVMapAny(c, "providerConfig"),
Subject: xmaps.GetString(c, "subject"),
Message: xmaps.GetString(c, "message"),
SkipOnAllPrevSkipped: xmaps.GetBool(c, "skipOnAllPrevSkipped"),
}
}
func (c WorkflowNodeConfig) AsBranchBlock() WorkflowNodeConfigForBranchBlock {
expression := c["expression"]
if expression == nil {
return WorkflowNodeConfigForBranchBlock{}
}
exprRaw, _ := json.Marshal(expression)
expr, err := expr.UnmarshalExpr([]byte(exprRaw))
if err != nil {
return WorkflowNodeConfigForBranchBlock{}
}
return WorkflowNodeConfigForBranchBlock{
Expression: expr,
}
}
type WorkflowNodeConfigForBizApply struct {
@@ -153,82 +234,3 @@ type WorkflowNodeConfigForBizNotify struct {
type WorkflowNodeConfigForBranchBlock struct {
Expression expr.Expr `json:"expression"` // 条件表达式
}
func (n *WorkflowNode) GetConfigForBizApply() WorkflowNodeConfigForBizApply {
return WorkflowNodeConfigForBizApply{
Domains: xmaps.GetString(n.Data.Config, "domains"),
ContactEmail: xmaps.GetString(n.Data.Config, "contactEmail"),
ChallengeType: xmaps.GetString(n.Data.Config, "challengeType"),
Provider: xmaps.GetString(n.Data.Config, "provider"),
ProviderAccessId: xmaps.GetString(n.Data.Config, "providerAccessId"),
ProviderConfig: xmaps.GetKVMapAny(n.Data.Config, "providerConfig"),
KeyAlgorithm: xmaps.GetOrDefaultString(n.Data.Config, "keyAlgorithm", string(CertificateKeyAlgorithmTypeRSA2048)),
CAProvider: xmaps.GetString(n.Data.Config, "caProvider"),
CAProviderAccessId: xmaps.GetString(n.Data.Config, "caProviderAccessId"),
CAProviderConfig: xmaps.GetKVMapAny(n.Data.Config, "caProviderConfig"),
ACMEProfile: xmaps.GetString(n.Data.Config, "acmeProfile"),
Nameservers: xmaps.GetString(n.Data.Config, "nameservers"),
DnsPropagationWait: xmaps.GetInt32(n.Data.Config, "dnsPropagationWait"),
DnsPropagationTimeout: xmaps.GetInt32(n.Data.Config, "dnsPropagationTimeout"),
DnsTTL: xmaps.GetInt32(n.Data.Config, "dnsTTL"),
DisableFollowCNAME: xmaps.GetBool(n.Data.Config, "disableFollowCNAME"),
DisableARI: xmaps.GetBool(n.Data.Config, "disableARI"),
SkipBeforeExpiryDays: xmaps.GetInt32(n.Data.Config, "skipBeforeExpiryDays"),
}
}
func (n *WorkflowNode) GetConfigForBizUpload() WorkflowNodeConfigForBizUpload {
return WorkflowNodeConfigForBizUpload{
Certificate: xmaps.GetString(n.Data.Config, "certificate"),
PrivateKey: xmaps.GetString(n.Data.Config, "privateKey"),
Domains: xmaps.GetString(n.Data.Config, "domains"),
}
}
func (n *WorkflowNode) GetConfigForBizMonitor() WorkflowNodeConfigForBizMonitor {
host := xmaps.GetString(n.Data.Config, "host")
return WorkflowNodeConfigForBizMonitor{
Host: host,
Port: xmaps.GetOrDefaultInt32(n.Data.Config, "port", 443),
Domain: xmaps.GetOrDefaultString(n.Data.Config, "domain", host),
RequestPath: xmaps.GetString(n.Data.Config, "path"),
}
}
func (n *WorkflowNode) GetConfigForBizDeploy() WorkflowNodeConfigForBizDeploy {
return WorkflowNodeConfigForBizDeploy{
CertificateOutputNodeId: xmaps.GetString(n.Data.Config, "certificateOutputNodeId"),
Provider: xmaps.GetString(n.Data.Config, "provider"),
ProviderAccessId: xmaps.GetString(n.Data.Config, "providerAccessId"),
ProviderConfig: xmaps.GetKVMapAny(n.Data.Config, "providerConfig"),
SkipOnLastSucceeded: xmaps.GetBool(n.Data.Config, "skipOnLastSucceeded"),
}
}
func (n *WorkflowNode) GetConfigForBizNotify() WorkflowNodeConfigForBizNotify {
return WorkflowNodeConfigForBizNotify{
Provider: xmaps.GetString(n.Data.Config, "provider"),
ProviderAccessId: xmaps.GetString(n.Data.Config, "providerAccessId"),
ProviderConfig: xmaps.GetKVMapAny(n.Data.Config, "providerConfig"),
Subject: xmaps.GetString(n.Data.Config, "subject"),
Message: xmaps.GetString(n.Data.Config, "message"),
SkipOnAllPrevSkipped: xmaps.GetBool(n.Data.Config, "skipOnAllPrevSkipped"),
}
}
func (n *WorkflowNode) GetConfigForBranchBlock() WorkflowNodeConfigForBranchBlock {
expression := n.Data.Config["expression"]
if expression == nil {
return WorkflowNodeConfigForBranchBlock{}
}
exprRaw, _ := json.Marshal(expression)
expr, err := expr.UnmarshalExpr([]byte(exprRaw))
if err != nil {
return WorkflowNodeConfigForBranchBlock{}
}
return WorkflowNodeConfigForBranchBlock{
Expression: expr,
}
}
internal/domain/workflow_output.go
+1
View File
@@ -7,6 +7,7 @@ type WorkflowOutput struct {
WorkflowId string `json:"workflowId" db:"workflowRef"`
RunId string `json:"runId" db:"runRef"`
NodeId string `json:"nodeId" db:"nodeId"`
NodeConfig WorkflowNodeConfig `json:"nodeConfig" db:"nodeConfig"`
Outputs []*WorkflowOutputEntry `json:"outputs" db:"outputs"`
Succeeded bool `json:"succeeded" db:"succeeded"`
}
internal/notify/notifier.go
+1 -1
View File
@@ -29,7 +29,7 @@ func NewWithWorkflowNode(config NotifierWithWorkflowNodeConfig) (Notifier, error
return nil, fmt.Errorf("the node type is '%s', expected '%s'", string(config.Node.Type), string(domain.WorkflowNodeTypeBizNotify))
}
nodeCfg := config.Node.GetConfigForBizNotify()
nodeCfg := config.Node.Data.Config.AsBizNotify()
options := &notifierProviderOptions{
Provider: domain.NotificationProviderType(nodeCfg.Provider),
ProviderAccessConfig: make(map[string]any),
internal/repository/workflow_output.go
+7
View File
@@ -107,6 +107,11 @@ func (r *WorkflowOutputRepository) castRecordToModel(record *core.Record) (*doma
return nil, fmt.Errorf("the record is nil")
}
nodeConfig := make(domain.WorkflowNodeConfig)
if err := record.UnmarshalJSONField("nodeConfig", &nodeConfig); err != nil {
return nil, fmt.Errorf("field 'nodeConfig' is malformed")
}
outputs := make([]*domain.WorkflowOutputEntry, 0)
if err := record.UnmarshalJSONField("outputs", &outputs); err != nil {
return nil, fmt.Errorf("field 'outputs' is malformed")
@@ -121,6 +126,7 @@ func (r *WorkflowOutputRepository) castRecordToModel(record *core.Record) (*doma
WorkflowId: record.GetString("workflowRef"),
RunId: record.GetString("runRef"),
NodeId: record.GetString("nodeId"),
NodeConfig: nodeConfig,
Outputs: outputs,
Succeeded: record.GetBool("succeeded"),
}
@@ -145,6 +151,7 @@ func (r *WorkflowOutputRepository) saveRecord(workflowOutput *domain.WorkflowOut
record.Set("workflowRef", workflowOutput.WorkflowId)
record.Set("runRef", workflowOutput.RunId)
record.Set("nodeId", workflowOutput.NodeId)
record.Set("nodeConfig", workflowOutput.NodeConfig)
record.Set("outputs", workflowOutput.Outputs)
record.Set("succeeded", workflowOutput.Succeeded)
if err := app.GetApp().Save(record); err != nil {
internal/workflow/engine/engine.go
+10 -7
View File
@@ -24,6 +24,8 @@ type WorkflowEngine interface {
}
type workflowEngine struct {
logger *slog.Logger
executorRegistry map[NodeType]NodeExecutor
hooksMtx sync.RWMutex
@@ -184,7 +186,7 @@ func (we *workflowEngine) fireOnStartHooks(ctx context.Context) {
defer we.hooksMtx.RUnlock()
for _, cb := range we.onStartHooks {
if cbErr := cb(ctx); cbErr != nil {
app.GetLogger().Error("workflow engine: error in onStart hook", slog.Any("error", cbErr))
we.logger.Error("workflow engine: error in onStart hook", slog.Any("error", cbErr))
}
}
}
@@ -194,7 +196,7 @@ func (we *workflowEngine) fireOnEndHooks(ctx context.Context) {
defer we.hooksMtx.RUnlock()
for _, cb := range we.onEndHooks {
if cbErr := cb(ctx); cbErr != nil {
app.GetLogger().Error("workflow engine: error in onEnd hook", slog.Any("error", cbErr))
we.logger.Error("workflow engine: error in onEnd hook", slog.Any("error", cbErr))
}
}
}
@@ -204,7 +206,7 @@ func (we *workflowEngine) fireOnErrorHooks(ctx context.Context, err error) {
defer we.hooksMtx.RUnlock()
for _, cb := range we.onErrorHooks {
if cbErr := cb(ctx, err); cbErr != nil {
app.GetLogger().Error("workflow engine: error in onError hook", slog.Any("error", cbErr))
we.logger.Error("workflow engine: error in onError hook", slog.Any("error", cbErr))
}
}
}
@@ -214,7 +216,7 @@ func (we *workflowEngine) fireOnNodeStartHooks(ctx context.Context, node *Node)
defer we.hooksMtx.RUnlock()
for _, cb := range we.onNodeStartHooks {
if cbErr := cb(ctx, node); cbErr != nil {
app.GetLogger().Error("workflow engine: error in onNodeStart hook", slog.Any("error", cbErr))
we.logger.Error("workflow engine: error in onNodeStart hook", slog.Any("error", cbErr))
}
}
}
@@ -224,7 +226,7 @@ func (we *workflowEngine) fireOnNodeEndHooks(ctx context.Context, node *Node, re
defer we.hooksMtx.RUnlock()
for _, cb := range we.onNodeEndHooks {
if cbErr := cb(ctx, node, result); cbErr != nil {
app.GetLogger().Error("workflow engine: error in onNodeEnd hook", slog.Any("error", cbErr))
we.logger.Error("workflow engine: error in onNodeEnd hook", slog.Any("error", cbErr))
}
}
}
@@ -234,7 +236,7 @@ func (we *workflowEngine) fireOnNodeErrorHooks(ctx context.Context, node *Node,
defer we.hooksMtx.RUnlock()
for _, cb := range we.onNodeErrorHooks {
if cbErr := cb(ctx, node, err); cbErr != nil {
app.GetLogger().Error("workflow engine: error in onNodeError hook", slog.Any("error", cbErr))
we.logger.Error("workflow engine: error in onNodeError hook", slog.Any("error", cbErr))
}
}
}
@@ -244,7 +246,7 @@ func (we *workflowEngine) fireOnNodeLoggingHooks(ctx context.Context, node *Node
defer we.hooksMtx.RUnlock()
for _, cb := range we.onNodeLoggingHooks {
if cbErr := cb(ctx, node, log); cbErr != nil {
app.GetLogger().Error("workflow engine: error in onNodeLogging hook", slog.Any("error", cbErr))
we.logger.Error("workflow engine: error in onNodeLogging hook", slog.Any("error", cbErr))
}
}
}
@@ -252,6 +254,7 @@ func (we *workflowEngine) fireOnNodeLoggingHooks(ctx context.Context, node *Node
func NewWorkflowEngine() WorkflowEngine {
engine := &workflowEngine{
executorRegistry: make(map[NodeType]NodeExecutor),
logger: app.GetLogger(),
}
engine.executorRegistry[NodeTypeStart] = newStartNodeExecutor()
engine.executorRegistry[NodeTypeEnd] = newEndNodeExecutor()
internal/workflow/engine/executor_bizapply.go
+4 -14
View File
@@ -16,14 +16,13 @@ type bizApplyNodeExecutor struct {
nodeExecutor
certificateRepo certificateRepository
wfrunRepo workflowRunRepository
wfoutputRepo workflowOutputRepository
}
func (ne *bizApplyNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeExecutionResult, error) {
execRes := &NodeExecutionResult{}
nodeCfg := execCtx.Node.GetConfigForBizApply()
nodeCfg := execCtx.Node.Data.Config.AsBizApply()
ne.logger.Info("ready to request certificate ...", slog.Any("config", nodeCfg))
// 查询上次执行结果
@@ -87,6 +86,7 @@ func (ne *bizApplyNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeExe
WorkflowId: execCtx.WorkflowId,
RunId: execCtx.RunId,
NodeId: execCtx.Node.Id,
NodeConfig: execCtx.Node.Data.Config,
Succeeded: true,
Outputs: []*domain.WorkflowOutputEntry{
{
@@ -136,20 +136,11 @@ func (ne *bizApplyNodeExecutor) getLastOutputArtifacts(execCtx *NodeExecutionCon
}
func (ne *bizApplyNodeExecutor) checkCanSkip(execCtx *NodeExecutionContext, lastOutput *domain.WorkflowOutput, lastCertificate *domain.Certificate) (_skip bool, _reason string) {
thisNodeCfg := execCtx.Node.GetConfigForBizApply()
thisNodeCfg := execCtx.Node.Data.Config.AsBizApply()
if lastOutput != nil && lastOutput.Succeeded {
lastRun, err := ne.wfrunRepo.GetById(execCtx.ctx, lastOutput.RunId)
if err != nil {
return true, "failed to get last run"
}
lastNode, lastNodeExists := lastRun.Graph.GetNodeById(lastOutput.NodeId)
if !lastNodeExists {
return true, "failed to get last run node"
}
// 比较和上次申请时的关键配置（即影响证书签发的）参数是否一致
lastNodeCfg := lastNode.GetConfigForBizApply()
lastNodeCfg := lastOutput.NodeConfig.AsBizApply()
if thisNodeCfg.Domains != lastNodeCfg.Domains {
return false, "the configuration item 'Domains' changed"
@@ -201,7 +192,6 @@ func newBizApplyNodeExecutor() NodeExecutor {
return &bizApplyNodeExecutor{
nodeExecutor: nodeExecutor{logger: slog.Default()},
certificateRepo: repository.NewCertificateRepository(),
wfrunRepo: repository.NewWorkflowRunRepository(),
wfoutputRepo: repository.NewWorkflowOutputRepository(),
}
}
internal/workflow/engine/executor_bizdeploy.go
+4 -14
View File
@@ -14,14 +14,13 @@ type bizDeployNodeExecutor struct {
nodeExecutor
certificateRepo certificateRepository
wfrunRepo workflowRunRepository
wfoutputRepo workflowOutputRepository
}
func (ne *bizDeployNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeExecutionResult, error) {
execRes := &NodeExecutionResult{}
nodeCfg := execCtx.Node.GetConfigForBizDeploy()
nodeCfg := execCtx.Node.Data.Config.AsBizDeploy()
ne.logger.Info("ready to deploy certificate ...", slog.Any("config", nodeCfg))
// 查询上次执行结果
@@ -74,6 +73,7 @@ func (ne *bizDeployNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeEx
WorkflowId: execCtx.WorkflowId,
RunId: execCtx.RunId,
NodeId: execCtx.Node.Id,
NodeConfig: execCtx.Node.Data.Config,
Succeeded: true,
}
if _, err := ne.wfoutputRepo.Save(execCtx.ctx, output); err != nil {
@@ -98,20 +98,11 @@ func (ne *bizDeployNodeExecutor) getLastOutputArtifacts(execCtx *NodeExecutionCo
}
func (ne *bizDeployNodeExecutor) checkCanSkip(execCtx *NodeExecutionContext, lastOutput *domain.WorkflowOutput) (_skip bool, _reason string) {
thisNodeCfg := execCtx.Node.GetConfigForBizDeploy()
thisNodeCfg := execCtx.Node.Data.Config.AsBizDeploy()
if lastOutput != nil && lastOutput.Succeeded {
lastRun, err := ne.wfrunRepo.GetById(execCtx.ctx, lastOutput.RunId)
if err != nil {
return true, "failed to get last run"
}
lastNode, lastNodeExists := lastRun.Graph.GetNodeById(lastOutput.NodeId)
if !lastNodeExists {
return true, "failed to get last run node"
}
// 比较和上次部署时的关键配置（即影响证书部署的）参数是否一致
lastNodeCfg := lastNode.GetConfigForBizDeploy()
lastNodeCfg := lastOutput.NodeConfig.AsBizDeploy()
if thisNodeCfg.ProviderAccessId != lastNodeCfg.ProviderAccessId {
return false, "the configuration item 'ProviderAccessId' changed"
@@ -132,7 +123,6 @@ func newBizDeployNodeExecutor() NodeExecutor {
return &bizDeployNodeExecutor{
nodeExecutor: nodeExecutor{logger: slog.Default()},
certificateRepo: repository.NewCertificateRepository(),
wfrunRepo: repository.NewWorkflowRunRepository(),
wfoutputRepo: repository.NewWorkflowOutputRepository(),
}
}
internal/workflow/engine/executor_bizmonitor.go
+1 -1
View File
@@ -25,7 +25,7 @@ type bizMonitorNodeExecutor struct {
func (ne *bizMonitorNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeExecutionResult, error) {
execRes := &NodeExecutionResult{}
nodeCfg := execCtx.Node.GetConfigForBizMonitor()
nodeCfg := execCtx.Node.Data.Config.AsBizMonitor()
ne.logger.Info("ready to monitor certificate ...", slog.Any("config", nodeCfg))
targetAddr := net.JoinHostPort(nodeCfg.Host, strconv.Itoa(int(nodeCfg.Port)))
internal/workflow/engine/executor_biznotify.go
+2 -2
View File
@@ -17,7 +17,7 @@ type bizNotifyNodeExecutor struct {
func (ne *bizNotifyNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeExecutionResult, error) {
execRes := &NodeExecutionResult{}
nodeCfg := execCtx.Node.GetConfigForBizNotify()
nodeCfg := execCtx.Node.Data.Config.AsBizNotify()
ne.logger.Info("ready to send notification ...", slog.Any("config", nodeCfg))
// 检测是否可以跳过本次执行
@@ -49,7 +49,7 @@ func (ne *bizNotifyNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeEx
}
func (ne *bizNotifyNodeExecutor) checkCanSkip(execCtx *NodeExecutionContext) (_skip bool) {
thisNodeCfg := execCtx.Node.GetConfigForBizNotify()
thisNodeCfg := execCtx.Node.Data.Config.AsBizNotify()
if !thisNodeCfg.SkipOnAllPrevSkipped {
return false
}
internal/workflow/engine/executor_bizupload.go
+4 -14
View File
@@ -14,14 +14,13 @@ type bizUploadNodeExecutor struct {
nodeExecutor
certificateRepo certificateRepository
wfrunRepo workflowRunRepository
wfoutputRepo workflowOutputRepository
}
func (ne *bizUploadNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeExecutionResult, error) {
execRes := &NodeExecutionResult{}
nodeCfg := execCtx.Node.GetConfigForBizUpload()
nodeCfg := execCtx.Node.Data.Config.AsBizUpload()
ne.logger.Info("ready to upload certiticate ...", slog.Any("config", nodeCfg))
// 查询上次执行结果
@@ -54,6 +53,7 @@ func (ne *bizUploadNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeEx
WorkflowId: execCtx.WorkflowId,
RunId: execCtx.RunId,
NodeId: execCtx.Node.Id,
NodeConfig: execCtx.Node.Data.Config,
Succeeded: true,
Outputs: []*domain.WorkflowOutputEntry{
{
@@ -97,20 +97,11 @@ func (ne *bizUploadNodeExecutor) getLastOutputArtifacts(execCtx *NodeExecutionCo
}
func (ne *bizUploadNodeExecutor) checkCanSkip(execCtx *NodeExecutionContext, lastOutput *domain.WorkflowOutput, lastCertificate *domain.Certificate) (_skip bool, _reason string) {
thisNodeCfg := execCtx.Node.GetConfigForBizUpload()
thisNodeCfg := execCtx.Node.Data.Config.AsBizUpload()
if lastOutput != nil && lastOutput.Succeeded {
lastRun, err := ne.wfrunRepo.GetById(execCtx.ctx, lastOutput.RunId)
if err != nil {
return true, "failed to get last run"
}
lastNode, lastNodeExists := lastRun.Graph.GetNodeById(lastOutput.NodeId)
if !lastNodeExists {
return true, "failed to get last run node"
}
// 比较和上次上传时的关键配置（即影响证书上传的）参数是否一致
lastNodeCfg := lastNode.GetConfigForBizUpload()
lastNodeCfg := lastOutput.NodeConfig.AsBizUpload()
if strings.TrimSpace(thisNodeCfg.Certificate) != strings.TrimSpace(lastNodeCfg.Certificate) {
return false, "the configuration item 'Certificate' changed"
@@ -135,7 +126,6 @@ func newBizUploadNodeExecutor() NodeExecutor {
return &bizUploadNodeExecutor{
nodeExecutor: nodeExecutor{logger: slog.Default()},
certificateRepo: repository.NewCertificateRepository(),
wfrunRepo: repository.NewWorkflowRunRepository(),
wfoutputRepo: repository.NewWorkflowOutputRepository(),
}
}
internal/workflow/engine/executor_condition.go
+2 -2
View File
@@ -42,7 +42,7 @@ func (ne *conditionNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeEx
}
if len(errs) > 0 {
return execRes, errors.Join(errs...)
return execRes, fmt.Errorf("error occurred when executing child nodes: %w", errors.Join(errs...))
}
return execRes, nil
@@ -61,7 +61,7 @@ type branchBlockNodeExecutor struct {
func (ne *branchBlockNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeExecutionResult, error) {
execRes := &NodeExecutionResult{}
nodeCfg := execCtx.Node.GetConfigForBranchBlock()
nodeCfg := execCtx.Node.Data.Config.AsBranchBlock()
if nodeCfg.Expression == nil {
ne.logger.Info("enter this branch without any conditions")
} else {
internal/workflow/engine/executor_trycatch.go
+3 -2
View File
@@ -2,6 +2,7 @@
import (
"errors"
"fmt"
"log/slog"
"github.com/samber/lo"
@@ -59,10 +60,10 @@ func (ne *tryCatchNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeExe
}
if len(catchErrs) > 0 {
return execRes, errors.Join(append(tryErrs, catchErrs...)...)
return execRes, fmt.Errorf("error occurred when executing child nodes: %w", errors.Join(append(tryErrs, catchErrs...)...))
}
return execRes, errors.Join(tryErrs...)
return execRes, fmt.Errorf("error occurred when executing child nodes: %w", errors.Join(tryErrs...))
}
return execRes, nil
migrations/1755187200_v0.4.0_migrate.go
+25 -5
View File
@@ -556,7 +556,7 @@ func init() {
// update collection `workflow_output`
// - rename field `workflowId` to `workflowRef`
// - rename field `runId` to `runRef`
// - delete field `node`
// - rename field `node` to `nodeConfig`
{
collection, err := app.FindCollectionByNameOrId("bqnxb95f2cooowp")
if err != nil {
@@ -604,8 +604,17 @@ func init() {
return err
}
if field := collection.Fields.GetByName("node"); field != nil {
collection.Fields.RemoveById(field.GetId())
if err := collection.Fields.AddMarshaledJSONAt(4, []byte(`{
"hidden": false,
"id": "json2239752261",
"maxSize": 5000000,
"name": "nodeConfig",
"presentable": false,
"required": false,
"system": false,
"type": "json"
}`)); err != nil {
return err
}
if err := app.Save(collection); err != nil {
@@ -1049,6 +1058,7 @@ func init() {
}
// update collection `workflow_output`
// - migrate field `nodeConfig`
// - migrate field `outputs`
{
collection, err := app.FindCollectionByNameOrId("bqnxb95f2cooowp")
@@ -1062,6 +1072,18 @@ func init() {
for _, record := range records {
changed := false
nodeConfig := make(map[string]any)
if err := record.UnmarshalJSONField("nodeConfig", &nodeConfig); err == nil {
if _, ok := nodeConfig["id"]; ok {
if _, ok := nodeConfig["type"]; ok {
if _, ok := nodeConfig["config"]; ok {
record.Set("nodeConfig", nodeConfig["config"])
changed = true
}
}
}
}
outputs := make([]map[string]any, 0)
if err := record.UnmarshalJSONField("outputs", &outputs); err == nil {
for i, output := range outputs {
@@ -1077,8 +1099,6 @@ func init() {
record.Set("outputs", outputs)
changed = true
}
} else {
println(err.Error())
}
if changed {