Feat: Commit step-generate data without success (#2539)

* Feat: commit without success

* Feat: add test case

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, "")

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
 }

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

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

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) {