diff --git a/pkg/workflow/context/context.go b/pkg/workflow/context/context.go index f59d310e9..107377f02 100644 --- a/pkg/workflow/context/context.go +++ b/pkg/workflow/context/context.go @@ -48,6 +48,7 @@ type WorkflowContext struct { store corev1.ConfigMap components map[string]*ComponentManifest vars *value.Value + modified bool } // GetComponent Get ComponentManifest from workflow context. @@ -70,7 +71,11 @@ func (wf *WorkflowContext) PatchComponent(name string, patchValue *value.Value) if err != nil { return err } - return component.Patch(patchValue) + if err := component.Patch(patchValue); err != nil { + return err + } + wf.modified = true + return nil } // GetVar get variable from workflow context. @@ -87,7 +92,11 @@ func (wf *WorkflowContext) SetVar(v *value.Value, paths ...string) error { if err := wf.vars.FillRaw(str, paths...); err != nil { return err } - return wf.vars.Error() + if err := wf.vars.Error(); err != nil { + return err + } + wf.modified = true + return nil } // MakeParameter make 'value' with interface{} @@ -106,6 +115,9 @@ func (wf *WorkflowContext) MakeParameter(parameter interface{}) (*value.Value, e // Commit the workflow context and persist it's content. func (wf *WorkflowContext) Commit() error { + if !wf.modified { + return nil + } if err := wf.writeToStore(); err != nil { return err } @@ -303,6 +315,7 @@ func newContext(cli client.Client, ns, app string) (*WorkflowContext, error) { cli: cli, store: store, components: map[string]*ComponentManifest{}, + modified: true, } var err error wfCtx.vars, err = value.NewValue("", nil, "") diff --git a/pkg/workflow/recorder/recorder.go b/pkg/workflow/recorder/recorder.go index 0d1dd5b9b..bbfd0e713 100644 --- a/pkg/workflow/recorder/recorder.go +++ b/pkg/workflow/recorder/recorder.go @@ -89,8 +89,12 @@ func (r *recorder) Save(version string, data []byte) Store { rv.Data = runtime.RawExtension{ Raw: data, } - if err := r.cli.Create(context.Background(), rv); err != nil && !kerrors.IsAlreadyExists(err) { - r.err = errors.WithMessagef(err, "save record %s/%s", rv.Namespace, rv.Name) + if err := r.cli.Create(context.Background(), rv); err != nil { + if kerrors.IsAlreadyExists(err) { + r.err = r.cli.Update(context.Background(), rv) + } else { + r.err = errors.WithMessagef(err, "save record %s/%s", rv.Namespace, rv.Name) + } } return r } diff --git a/pkg/workflow/recorder/recorder_test.go b/pkg/workflow/recorder/recorder_test.go index d1c0b1d06..539c3a779 100644 --- a/pkg/workflow/recorder/recorder_test.go +++ b/pkg/workflow/recorder/recorder_test.go @@ -24,6 +24,7 @@ import ( "github.com/pkg/errors" "gotest.tools/assert" apps "k8s.io/api/apps/v1" + kerrors "k8s.io/apimachinery/pkg/api/errors" "sigs.k8s.io/controller-runtime/pkg/client" "github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1" @@ -47,6 +48,11 @@ func TestRecord(t *testing.T) { assert.Equal(t, crs.Items[0].Name, "record-test-app-v2") assert.Equal(t, crs.Items[1].Name, "record-test-app-v3") + + // check update old recorder. + err = With(cli, app).Save("v3", data).Error() + assert.NilError(t, err) + creatErrorEnable = true err = With(cli, app).Save("v1", data).Error() assert.Equal(t, err.Error(), "save record default/record-test-app-v1: mock create error") @@ -86,12 +92,29 @@ func makeMockClient() client.Client { } return nil }, + MockUpdate: func(ctx context.Context, obj client.Object, opts ...client.UpdateOption) error { + o, ok := obj.(*apps.ControllerRevision) + if ok { + for index, item := range items { + if item.Name == o.Name && item.Namespace == o.Namespace { + items[index] = *o + return nil + } + } + } + return kerrors.NewNotFound(apps.Resource("ControllerRevision"), o.Name) + }, MockCreate: func(ctx context.Context, obj client.Object, opts ...client.CreateOption) error { if creatErrorEnable { return errors.New("mock create error") } o, ok := obj.(*apps.ControllerRevision) if ok { + for _, item := range items { + if item.Name == o.Name && item.Namespace == o.Namespace { + return kerrors.NewAlreadyExists(apps.Resource("ControllerRevision"), o.Name) + } + } items = append(items, *o) } return nil diff --git a/pkg/workflow/workflow.go b/pkg/workflow/workflow.go index 3bc279263..e3fd34cec 100644 --- a/pkg/workflow/workflow.go +++ b/pkg/workflow/workflow.go @@ -270,6 +270,10 @@ func (e *engine) steps(wfCtx wfContext.Context, taskRunners []wfTypes.TaskRunner e.updateStepStatus(status) + if err := wfCtx.Commit(); err != nil { + return errors.WithMessage(err, "commit workflow context") + } + if status.Phase != common.WorkflowStepPhaseSucceeded { if e.isDag() { continue @@ -277,10 +281,6 @@ func (e *engine) steps(wfCtx wfContext.Context, taskRunners []wfTypes.TaskRunner return nil } - if err := wfCtx.Commit(); err != nil { - return errors.WithMessage(err, "commit workflow context") - } - e.finishStep(operation) if e.needStop() { return nil diff --git a/pkg/workflow/workflow_test.go b/pkg/workflow/workflow_test.go index bfa2f2317..e1635a3b3 100644 --- a/pkg/workflow/workflow_test.go +++ b/pkg/workflow/workflow_test.go @@ -20,6 +20,8 @@ import ( "context" "encoding/json" + "github.com/oam-dev/kubevela/pkg/cue/model/value" + . "github.com/onsi/ginkgo" . "github.com/onsi/gomega" @@ -344,6 +346,31 @@ var _ = Describe("Test Workflow", func() { }}, })).Should(BeEquivalentTo("")) }) + + It("step commit data without success", func() { + app, runners := makeTestCase([]oamcore.WorkflowStep{ + { + Name: "s1", + Type: "wait-with-set-var", + }, + { + Name: "s2", + Type: "success", + }, + }) + wf := NewWorkflow(app, k8sClient, common.WorkflowModeStep) + state, err := wf.ExecuteSteps(context.Background(), revision, runners) + Expect(err).ToNot(HaveOccurred()) + Expect(state).Should(BeEquivalentTo(common.WorkflowStateExecuting)) + Expect(app.Status.Workflow.Steps[0].Phase).Should(BeEquivalentTo(common.WorkflowStepPhaseRunning)) + wfCtx, err := wfContext.LoadContext(k8sClient, app.Namespace, app.Name) + Expect(err).ToNot(HaveOccurred()) + v, err := wfCtx.GetVar("saved") + Expect(err).ToNot(HaveOccurred()) + saved, err := v.CueValue().Bool() + Expect(err).ToNot(HaveOccurred()) + Expect(saved).Should(BeEquivalentTo(true)) + }) }) func makeTestCase(steps []oamcore.WorkflowStep) (*oamcore.Application, []wfTypes.TaskRunner) { @@ -413,6 +440,16 @@ func makeRunner(name string, tpy string) wfTypes.TaskRunner { Phase: common.WorkflowStepPhaseRunning, }, &wfTypes.Operation{}, errors.New("error for test") } + case "wait-with-set-var": + run = func(ctx wfContext.Context, options *wfTypes.TaskRunOptions) (common.WorkflowStepStatus, *wfTypes.Operation, error) { + v, _ := value.NewValue(`saved: true`, nil, "") + err := ctx.SetVar(v) + return common.WorkflowStepStatus{ + Name: name, + Type: "wait-with-set-var", + Phase: common.WorkflowStepPhaseRunning, + }, &wfTypes.Operation{}, err + } default: run = func(ctx wfContext.Context, options *wfTypes.TaskRunOptions) (common.WorkflowStepStatus, *wfTypes.Operation, error) {