diff --git a/apis/core.oam.dev/common/types.go b/apis/core.oam.dev/common/types.go index 01ba8cf74..cadb6a97b 100644 --- a/apis/core.oam.dev/common/types.go +++ b/apis/core.oam.dev/common/types.go @@ -216,19 +216,19 @@ type WorkflowState string const ( // WorkflowStateInitializing means the workflow is in initial state - WorkflowStateInitializing WorkflowState = "initializing" + WorkflowStateInitializing WorkflowState = "Initializing" // WorkflowStateTerminated means workflow is terminated manually, and it won't be started unless the spec changed. - WorkflowStateTerminated WorkflowState = "terminated" + WorkflowStateTerminated WorkflowState = "Terminated" // WorkflowStateSuspended means workflow is suspended manually, and it can be resumed. - WorkflowStateSuspended WorkflowState = "suspended" + WorkflowStateSuspended WorkflowState = "Suspended" // WorkflowStateSucceeded means workflow is running successfully, all steps finished. WorkflowStateSucceeded WorkflowState = "Succeeded" // WorkflowStateFinished means workflow is end. - WorkflowStateFinished WorkflowState = "finished" + WorkflowStateFinished WorkflowState = "Finished" // WorkflowStateExecuting means workflow is still running or waiting some steps. - WorkflowStateExecuting WorkflowState = "executing" + WorkflowStateExecuting WorkflowState = "Executing" // WorkflowStateSkipping means it will skip this reconcile and let next reconcile to handle it. - WorkflowStateSkipping WorkflowState = "skipping" + WorkflowStateSkipping WorkflowState = "Skipping" ) // ApplicationComponentStatus record the health status of App component diff --git a/cmd/core/main.go b/cmd/core/main.go index bee60cc9b..c0fbacfbe 100644 --- a/cmd/core/main.go +++ b/cmd/core/main.go @@ -54,8 +54,7 @@ import ( "github.com/oam-dev/kubevela/pkg/utils/system" "github.com/oam-dev/kubevela/pkg/utils/util" oamwebhook "github.com/oam-dev/kubevela/pkg/webhook/core.oam.dev" - "github.com/oam-dev/kubevela/pkg/workflow" - "github.com/oam-dev/kubevela/pkg/workflow/tasks/custom" + wfTypes "github.com/oam-dev/kubevela/pkg/workflow/types" "github.com/oam-dev/kubevela/version" ) @@ -141,9 +140,9 @@ func main() { standardcontroller.AddOptimizeFlags() standardcontroller.AddAdmissionFlags() flag.IntVar(&resourcekeeper.MaxDispatchConcurrent, "max-dispatch-concurrent", 10, "Set the max dispatch concurrent number, default is 10") - flag.IntVar(&workflow.MaxWorkflowWaitBackoffTime, "max-workflow-wait-backoff-time", 60, "Set the max workflow wait backoff time, default is 60") - flag.IntVar(&workflow.MaxWorkflowFailedBackoffTime, "max-workflow-failed-backoff-time", 300, "Set the max workflow wait backoff time, default is 300") - flag.IntVar(&custom.MaxWorkflowStepErrorRetryTimes, "max-workflow-step-error-retry-times", 10, "Set the max workflow step error retry times, default is 10") + flag.IntVar(&wfTypes.MaxWorkflowWaitBackoffTime, "max-workflow-wait-backoff-time", 60, "Set the max workflow wait backoff time, default is 60") + flag.IntVar(&wfTypes.MaxWorkflowFailedBackoffTime, "max-workflow-failed-backoff-time", 300, "Set the max workflow wait backoff time, default is 300") + flag.IntVar(&wfTypes.MaxWorkflowStepErrorRetryTimes, "max-workflow-step-error-retry-times", 10, "Set the max workflow step error retry times, default is 10") utilfeature.DefaultMutableFeatureGate.AddFlag(flag.CommandLine) flag.Parse() diff --git a/pkg/apiserver/domain/service/workflow.go b/pkg/apiserver/domain/service/workflow.go index 386e914e0..5a97ec1e5 100644 --- a/pkg/apiserver/domain/service/workflow.go +++ b/pkg/apiserver/domain/service/workflow.go @@ -43,7 +43,6 @@ import ( "github.com/oam-dev/kubevela/pkg/oam/util" utils2 "github.com/oam-dev/kubevela/pkg/utils" "github.com/oam-dev/kubevela/pkg/utils/apply" - "github.com/oam-dev/kubevela/pkg/workflow/tasks/custom" wfTypes "github.com/oam-dev/kubevela/pkg/workflow/types" ) @@ -629,23 +628,23 @@ func TerminateWorkflow(ctx context.Context, kubecli client.Client, app *v1beta1. for i, step := range steps { switch step.Phase { case common.WorkflowStepPhaseFailed: - if step.Reason != custom.StatusReasonFailedAfterRetries && step.Reason != custom.StatusReasonTimeout { - steps[i].Reason = custom.StatusReasonTerminate + if step.Reason != wfTypes.StatusReasonFailedAfterRetries && step.Reason != wfTypes.StatusReasonTimeout { + steps[i].Reason = wfTypes.StatusReasonTerminate } case common.WorkflowStepPhaseRunning: steps[i].Phase = common.WorkflowStepPhaseFailed - steps[i].Reason = custom.StatusReasonTerminate + steps[i].Reason = wfTypes.StatusReasonTerminate default: } for j, sub := range step.SubStepsStatus { switch sub.Phase { case common.WorkflowStepPhaseFailed: - if sub.Reason != custom.StatusReasonFailedAfterRetries && sub.Reason != custom.StatusReasonTimeout { - steps[i].SubStepsStatus[j].Phase = custom.StatusReasonTerminate + if sub.Reason != wfTypes.StatusReasonFailedAfterRetries && sub.Reason != wfTypes.StatusReasonTimeout { + steps[i].SubStepsStatus[j].Phase = wfTypes.StatusReasonTerminate } case common.WorkflowStepPhaseRunning: steps[i].SubStepsStatus[j].Phase = common.WorkflowStepPhaseFailed - steps[i].SubStepsStatus[j].Reason = custom.StatusReasonTerminate + steps[i].SubStepsStatus[j].Reason = wfTypes.StatusReasonTerminate default: } } diff --git a/pkg/controller/core.oam.dev/v1alpha2/application/application_controller_test.go b/pkg/controller/core.oam.dev/v1alpha2/application/application_controller_test.go index 9cd662dc9..c71fe5631 100644 --- a/pkg/controller/core.oam.dev/v1alpha2/application/application_controller_test.go +++ b/pkg/controller/core.oam.dev/v1alpha2/application/application_controller_test.go @@ -60,7 +60,7 @@ import ( common2 "github.com/oam-dev/kubevela/pkg/utils/common" "github.com/oam-dev/kubevela/pkg/workflow" "github.com/oam-dev/kubevela/pkg/workflow/debug" - "github.com/oam-dev/kubevela/pkg/workflow/tasks/custom" + wfTypes "github.com/oam-dev/kubevela/pkg/workflow/types" ) // TODO: Refactor the tests to not copy and paste duplicated code 10 times @@ -1944,7 +1944,7 @@ var _ = Describe("Test Application Controller", func() { Expect(checkApp.Status.Workflow.Message).Should(BeEquivalentTo(workflow.MessageInitializingWorkflow)) By("verify the first ten reconciles") - for i := 0; i < custom.MaxWorkflowStepErrorRetryTimes; i++ { + for i := 0; i < wfTypes.MaxWorkflowStepErrorRetryTimes; i++ { testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey}) Expect(k8sClient.Get(ctx, appKey, checkApp)).Should(BeNil()) Expect(checkApp.Status.Phase).Should(BeEquivalentTo(common.ApplicationRunningWorkflow)) @@ -1958,7 +1958,7 @@ var _ = Describe("Test Application Controller", func() { Expect(checkApp.Status.Phase).Should(BeEquivalentTo(common.ApplicationWorkflowSuspending)) Expect(checkApp.Status.Workflow.Message).Should(BeEquivalentTo(workflow.MessageSuspendFailedAfterRetries)) Expect(checkApp.Status.Workflow.Steps[1].Phase).Should(BeEquivalentTo(common.WorkflowStepPhaseFailed)) - Expect(checkApp.Status.Workflow.Steps[1].Reason).Should(BeEquivalentTo(custom.StatusReasonFailedAfterRetries)) + Expect(checkApp.Status.Workflow.Steps[1].Reason).Should(BeEquivalentTo(wfTypes.StatusReasonFailedAfterRetries)) By("resume the suspended application") Expect(k8sClient.Get(ctx, appKey, checkApp)).Should(BeNil()) @@ -1990,7 +1990,7 @@ var _ = Describe("Test Application Controller", func() { Expect(k8sClient.Get(ctx, appKey, checkApp)).Should(BeNil()) Expect(checkApp.Status.Phase).Should(BeEquivalentTo(common.ApplicationRunningWorkflow)) - for i := 0; i < custom.MaxWorkflowStepErrorRetryTimes-1; i++ { + for i := 0; i < wfTypes.MaxWorkflowStepErrorRetryTimes-1; i++ { testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey}) Expect(k8sClient.Get(ctx, appKey, checkApp)).Should(BeNil()) Expect(checkApp.Status.Phase).Should(BeEquivalentTo(common.ApplicationRunningWorkflow)) @@ -2005,7 +2005,7 @@ var _ = Describe("Test Application Controller", func() { Expect(checkApp.Status.Workflow.Message).Should(BeEquivalentTo(string(common.WorkflowStateExecuting))) Expect(checkApp.Status.Workflow.Steps[0].Phase).Should(BeEquivalentTo(common.WorkflowStepPhaseRunning)) Expect(checkApp.Status.Workflow.Steps[1].Phase).Should(BeEquivalentTo(common.WorkflowStepPhaseFailed)) - Expect(checkApp.Status.Workflow.Steps[1].Reason).Should(BeEquivalentTo(custom.StatusReasonFailedAfterRetries)) + Expect(checkApp.Status.Workflow.Steps[1].Reason).Should(BeEquivalentTo(wfTypes.StatusReasonFailedAfterRetries)) }) It("application with step by step workflow failed after retries", func() { @@ -2067,7 +2067,7 @@ var _ = Describe("Test Application Controller", func() { Expect(checkApp.Status.Workflow.Message).Should(BeEquivalentTo(workflow.MessageInitializingWorkflow)) By("verify the first twenty reconciles") - for i := 0; i < custom.MaxWorkflowStepErrorRetryTimes; i++ { + for i := 0; i < wfTypes.MaxWorkflowStepErrorRetryTimes; i++ { testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey}) Expect(k8sClient.Get(ctx, appKey, checkApp)).Should(BeNil()) Expect(checkApp.Status.Phase).Should(BeEquivalentTo(common.ApplicationRunningWorkflow)) @@ -2081,7 +2081,7 @@ var _ = Describe("Test Application Controller", func() { Expect(checkApp.Status.Phase).Should(BeEquivalentTo(common.ApplicationWorkflowSuspending)) Expect(checkApp.Status.Workflow.Message).Should(BeEquivalentTo(workflow.MessageSuspendFailedAfterRetries)) Expect(checkApp.Status.Workflow.Steps[1].Phase).Should(BeEquivalentTo(common.WorkflowStepPhaseFailed)) - Expect(checkApp.Status.Workflow.Steps[1].Reason).Should(BeEquivalentTo(custom.StatusReasonFailedAfterRetries)) + Expect(checkApp.Status.Workflow.Steps[1].Reason).Should(BeEquivalentTo(wfTypes.StatusReasonFailedAfterRetries)) By("resume the suspended application") Expect(k8sClient.Get(ctx, appKey, checkApp)).Should(BeNil()) @@ -2276,7 +2276,7 @@ var _ = Describe("Test Application Controller", func() { testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey}) By("verify the first ten reconciles") - for i := 0; i < custom.MaxWorkflowStepErrorRetryTimes; i++ { + for i := 0; i < wfTypes.MaxWorkflowStepErrorRetryTimes; i++ { testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey}) } @@ -2399,7 +2399,7 @@ var _ = Describe("Test Application Controller", func() { testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey}) By("verify the first ten reconciles") - for i := 0; i < custom.MaxWorkflowStepErrorRetryTimes; i++ { + for i := 0; i < wfTypes.MaxWorkflowStepErrorRetryTimes; i++ { testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey}) } @@ -2440,6 +2440,253 @@ var _ = Describe("Test Application Controller", func() { Expect(checkApp.Status.Phase).Should(BeEquivalentTo(common.ApplicationWorkflowTerminated)) }) + It("application with if expressions in workflow", func() { + ns := corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{ + Name: "app-with-if-expressions", + }, + } + Expect(k8sClient.Create(ctx, &ns)).Should(BeNil()) + healthComponentDef := &v1beta1.ComponentDefinition{} + hCDefJson, _ := yaml.YAMLToJSON([]byte(cdDefWithHealthStatusYaml)) + Expect(json.Unmarshal(hCDefJson, healthComponentDef)).Should(BeNil()) + healthComponentDef.Name = "worker-with-health" + healthComponentDef.Namespace = "app-with-if-expressions" + Expect(k8sClient.Create(ctx, healthComponentDef)).Should(BeNil()) + app := &v1beta1.Application{ + TypeMeta: metav1.TypeMeta{ + Kind: "Application", + APIVersion: "core.oam.dev/v1beta1", + }, + ObjectMeta: metav1.ObjectMeta{ + Name: "app-with-if-expressions", + Namespace: "app-with-if-expressions", + }, + Spec: v1beta1.ApplicationSpec{ + Components: []common.ApplicationComponent{ + { + Name: "myweb1", + Type: "worker-with-health", + Properties: &runtime.RawExtension{Raw: []byte(`{"cmd":["sleep","1000"],"image":"busybox","lives": "i am lives","enemies": "empty"}`)}, + }, + { + Name: "myweb2", + Type: "worker", + Properties: &runtime.RawExtension{Raw: []byte(`{"cmd":["sleep","1000"],"image":"busybox"}`)}, + }, + }, + Workflow: &v1beta1.Workflow{ + Steps: []v1beta1.WorkflowStep{ + { + Name: "suspend", + Type: "suspend", + Timeout: "1s", + Outputs: common.StepOutputs{ + { + Name: "suspend_output", + ValueFrom: "context.name", + }, + }, + }, + { + Name: "myweb1", + Type: "apply-component", + Inputs: common.StepInputs{ + { + From: "suspend_output", + ParameterKey: "", + }, + }, + If: `status.suspend.timeout && inputs.suspend_output == "app-with-if-expressions"`, + Properties: &runtime.RawExtension{Raw: []byte(`{"component":"myweb1"}`)}, + }, + { + Name: "myweb2", + If: "status.suspend.succeeded", + Type: "apply-component", + Properties: &runtime.RawExtension{Raw: []byte(`{"component":"myweb2"}`)}, + }, + }, + }, + }, + } + + Expect(k8sClient.Create(context.Background(), app)).Should(BeNil()) + appKey := types.NamespacedName{Namespace: ns.Name, Name: app.Name} + testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey}) + testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey}) + testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey}) + + checkApp := &v1beta1.Application{} + Expect(k8sClient.Get(ctx, appKey, checkApp)).Should(BeNil()) + Expect(checkApp.Status.Phase).Should(BeEquivalentTo(common.ApplicationWorkflowSuspending)) + + expDeployment := &v1.Deployment{} + web1Key := types.NamespacedName{Namespace: ns.Name, Name: "myweb1"} + Expect(k8sClient.Get(ctx, web1Key, expDeployment)).Should(util.NotFoundMatcher{}) + web2Key := types.NamespacedName{Namespace: ns.Name, Name: "myweb2"} + Expect(k8sClient.Get(ctx, web2Key, expDeployment)).Should(util.NotFoundMatcher{}) + + time.Sleep(time.Second) + testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey}) + Expect(k8sClient.Get(ctx, web1Key, expDeployment)).Should(BeNil()) + Expect(k8sClient.Get(ctx, web2Key, expDeployment)).Should(util.NotFoundMatcher{}) + + expDeployment.Status.Replicas = 1 + expDeployment.Status.ReadyReplicas = 1 + Expect(k8sClient.Status().Update(ctx, expDeployment)).Should(BeNil()) + + testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey}) + + Expect(k8sClient.Get(ctx, web2Key, expDeployment)).Should(util.NotFoundMatcher{}) + + Expect(k8sClient.Get(ctx, appKey, checkApp)).Should(BeNil()) + Expect(checkApp.Status.Phase).Should(BeEquivalentTo(common.ApplicationWorkflowTerminated)) + }) + + It("application with if expressions in workflow sub steps", func() { + ns := corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{ + Name: "app-with-if-expressions-workflow-sub-steps", + }, + } + Expect(k8sClient.Create(ctx, &ns)).Should(BeNil()) + healthComponentDef := &v1beta1.ComponentDefinition{} + hCDefJson, _ := yaml.YAMLToJSON([]byte(cdDefWithHealthStatusYaml)) + Expect(json.Unmarshal(hCDefJson, healthComponentDef)).Should(BeNil()) + healthComponentDef.Name = "worker-with-health" + healthComponentDef.Namespace = "app-with-if-expressions-workflow-sub-steps" + Expect(k8sClient.Create(ctx, healthComponentDef)).Should(BeNil()) + app := &v1beta1.Application{ + TypeMeta: metav1.TypeMeta{ + Kind: "Application", + APIVersion: "core.oam.dev/v1beta1", + }, + ObjectMeta: metav1.ObjectMeta{ + Name: "app-with-if-expressions-workflow-sub-steps", + Namespace: "app-with-if-expressions-workflow-sub-steps", + }, + Spec: v1beta1.ApplicationSpec{ + Components: []common.ApplicationComponent{ + { + Name: "myweb1", + Type: "worker-with-health", + Properties: &runtime.RawExtension{Raw: []byte(`{"cmd":["sleep","1000"],"image":"busybox","lives": "i am lives","enemies": "empty"}`)}, + }, + { + Name: "myweb1-sub", + Type: "worker-with-health", + Properties: &runtime.RawExtension{Raw: []byte(`{"cmd":["sleep","1000"],"image":"busybox","lives": "i am lives","enemies": "empty"}`)}, + }, + { + Name: "myweb2", + Type: "worker-with-health", + Properties: &runtime.RawExtension{Raw: []byte(`{"cmd":["sleep","1000"],"image":"busybox","lives": "i am lives","enemies": "empty"}`)}, + }, + { + Name: "myweb3", + Type: "worker", + Properties: &runtime.RawExtension{Raw: []byte(`{"cmd":["sleep","1000"],"image":"busybox"}`)}, + }, + }, + Workflow: &v1beta1.Workflow{ + Steps: []v1beta1.WorkflowStep{ + { + Name: "myweb1", + Type: "step-group", + SubSteps: []common.WorkflowSubStep{ + { + Name: "myweb1_sub1", + Type: "apply-component", + If: "status.myweb1_sub2.timeout", + DependsOn: []string{"myweb1_sub2"}, + Properties: &runtime.RawExtension{Raw: []byte(`{"component":"myweb1"}`)}, + }, + { + Name: "myweb1_sub2", + Type: "suspend", + Properties: &runtime.RawExtension{Raw: []byte(`{"duration":"1s"}`)}, + Outputs: common.StepOutputs{ + { + Name: "suspend_output", + ValueFrom: "context.name", + }, + }, + }, + { + Name: "myweb1_sub3", + Type: "apply-component", + DependsOn: []string{"myweb1_sub1"}, + Inputs: common.StepInputs{ + { + From: "suspend_output", + ParameterKey: "", + }, + }, + If: `status.myweb1_sub1.timeout || inputs.suspend_output == "app-with-if-expressions-workflow-sub-steps"`, + Properties: &runtime.RawExtension{Raw: []byte(`{"component":"myweb1-sub"}`)}, + }, + }, + }, + { + Name: "myweb2", + Type: "apply-component", + If: "status.myweb1.failed", + Properties: &runtime.RawExtension{Raw: []byte(`{"component":"myweb2"}`)}, + }, + { + Name: "myweb3", + Type: "apply-component", + Properties: &runtime.RawExtension{Raw: []byte(`{"component":"myweb3"}`)}, + }, + }, + }, + }, + } + + Expect(k8sClient.Create(context.Background(), app)).Should(BeNil()) + appKey := types.NamespacedName{Namespace: ns.Name, Name: app.Name} + testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey}) + testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey}) + testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey}) + + checkApp := &v1beta1.Application{} + Expect(k8sClient.Get(ctx, appKey, checkApp)).Should(BeNil()) + Expect(checkApp.Status.Phase).Should(BeEquivalentTo(common.ApplicationWorkflowSuspending)) + + expDeployment := &v1.Deployment{} + web1Key := types.NamespacedName{Namespace: ns.Name, Name: "myweb1"} + Expect(k8sClient.Get(ctx, web1Key, expDeployment)).Should(util.NotFoundMatcher{}) + web1SubKey := types.NamespacedName{Namespace: ns.Name, Name: "myweb1-sub"} + Expect(k8sClient.Get(ctx, web1SubKey, expDeployment)).Should(util.NotFoundMatcher{}) + web2Key := types.NamespacedName{Namespace: ns.Name, Name: "myweb2"} + Expect(k8sClient.Get(ctx, web2Key, expDeployment)).Should(util.NotFoundMatcher{}) + web3Key := types.NamespacedName{Namespace: ns.Name, Name: "myweb3"} + Expect(k8sClient.Get(ctx, web3Key, expDeployment)).Should(util.NotFoundMatcher{}) + + time.Sleep(time.Second) + testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey}) + + Expect(k8sClient.Get(ctx, appKey, checkApp)).Should(BeNil()) + Expect(checkApp.Status.Phase).Should(BeEquivalentTo(common.ApplicationRunningWorkflow)) + Expect(k8sClient.Get(ctx, web1Key, expDeployment)).Should(util.NotFoundMatcher{}) + Expect(k8sClient.Get(ctx, web1SubKey, expDeployment)).Should(BeNil()) + Expect(k8sClient.Get(ctx, web2Key, expDeployment)).Should(util.NotFoundMatcher{}) + Expect(k8sClient.Get(ctx, web3Key, expDeployment)).Should(util.NotFoundMatcher{}) + + expDeployment.Status.Replicas = 1 + expDeployment.Status.ReadyReplicas = 1 + Expect(k8sClient.Status().Update(ctx, expDeployment)).Should(BeNil()) + + testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey}) + testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey}) + + Expect(k8sClient.Get(ctx, web2Key, expDeployment)).Should(util.NotFoundMatcher{}) + Expect(k8sClient.Get(ctx, web3Key, expDeployment)).Should(BeNil()) + Expect(k8sClient.Get(ctx, appKey, checkApp)).Should(BeNil()) + Expect(checkApp.Status.Phase).Should(BeEquivalentTo(common.ApplicationRunning)) + }) + It("application with timeout in workflow", func() { ns := corev1.Namespace{ ObjectMeta: metav1.ObjectMeta{ @@ -2531,9 +2778,9 @@ var _ = Describe("Test Application Controller", func() { checkApp := &v1beta1.Application{} Expect(k8sClient.Get(ctx, appKey, checkApp)).Should(BeNil()) - Expect(checkApp.Status.Workflow.Steps[0].Reason).Should(Equal(custom.StatusReasonTimeout)) + Expect(checkApp.Status.Workflow.Steps[0].Reason).Should(Equal(wfTypes.StatusReasonTimeout)) Expect(checkApp.Status.Workflow.Steps[1].Phase).Should(Equal(common.WorkflowStepPhaseSucceeded)) - Expect(checkApp.Status.Workflow.Steps[2].Reason).Should(Equal(custom.StatusReasonSkip)) + Expect(checkApp.Status.Workflow.Steps[2].Reason).Should(Equal(wfTypes.StatusReasonSkip)) Expect(checkApp.Status.Phase).Should(BeEquivalentTo(common.ApplicationWorkflowTerminated)) }) @@ -2624,9 +2871,9 @@ var _ = Describe("Test Application Controller", func() { Expect(k8sClient.Get(ctx, web2Key, expDeployment)).Should(util.NotFoundMatcher{}) Expect(k8sClient.Get(ctx, appKey, checkApp)).Should(BeNil()) - Expect(checkApp.Status.Workflow.Steps[0].Reason).Should(Equal(custom.StatusReasonTimeout)) + Expect(checkApp.Status.Workflow.Steps[0].Reason).Should(Equal(wfTypes.StatusReasonTimeout)) Expect(checkApp.Status.Workflow.Steps[1].Phase).Should(Equal(common.WorkflowStepPhaseSucceeded)) - Expect(checkApp.Status.Workflow.Steps[2].Reason).Should(Equal(custom.StatusReasonSkip)) + Expect(checkApp.Status.Workflow.Steps[2].Reason).Should(Equal(wfTypes.StatusReasonSkip)) Expect(checkApp.Status.Phase).Should(BeEquivalentTo(common.ApplicationWorkflowTerminated)) }) @@ -2763,9 +3010,9 @@ var _ = Describe("Test Application Controller", func() { checkApp := &v1beta1.Application{} Expect(k8sClient.Get(ctx, appKey, checkApp)).Should(BeNil()) - Expect(checkApp.Status.Workflow.Steps[0].Reason).Should(Equal(custom.StatusReasonTimeout)) + Expect(checkApp.Status.Workflow.Steps[0].Reason).Should(Equal(wfTypes.StatusReasonTimeout)) Expect(checkApp.Status.Workflow.Steps[1].Phase).Should(Equal(common.WorkflowStepPhaseSucceeded)) - Expect(checkApp.Status.Workflow.Steps[2].Reason).Should(Equal(custom.StatusReasonSkip)) + Expect(checkApp.Status.Workflow.Steps[2].Reason).Should(Equal(wfTypes.StatusReasonSkip)) Expect(checkApp.Status.Phase).Should(BeEquivalentTo(common.ApplicationWorkflowTerminated)) }) @@ -2875,9 +3122,9 @@ var _ = Describe("Test Application Controller", func() { Expect(k8sClient.Get(ctx, web2Key, expDeployment)).Should(util.NotFoundMatcher{}) Expect(k8sClient.Get(ctx, appKey, checkApp)).Should(BeNil()) - Expect(checkApp.Status.Workflow.Steps[0].Reason).Should(Equal(custom.StatusReasonTimeout)) - Expect(checkApp.Status.Workflow.Steps[1].Reason).Should(Equal(custom.StatusReasonTimeout)) - Expect(checkApp.Status.Workflow.Steps[2].Reason).Should(Equal(custom.StatusReasonSkip)) + Expect(checkApp.Status.Workflow.Steps[0].Reason).Should(Equal(wfTypes.StatusReasonTimeout)) + Expect(checkApp.Status.Workflow.Steps[1].Reason).Should(Equal(wfTypes.StatusReasonTimeout)) + Expect(checkApp.Status.Workflow.Steps[2].Reason).Should(Equal(wfTypes.StatusReasonSkip)) Expect(checkApp.Status.Phase).Should(BeEquivalentTo(common.ApplicationWorkflowTerminated)) }) diff --git a/pkg/controller/core.oam.dev/v1alpha2/application/generator.go b/pkg/controller/core.oam.dev/v1alpha2/application/generator.go index 68c3a6917..d7c340fd8 100644 --- a/pkg/controller/core.oam.dev/v1alpha2/application/generator.go +++ b/pkg/controller/core.oam.dev/v1alpha2/application/generator.go @@ -33,6 +33,7 @@ import ( "github.com/oam-dev/kubevela/pkg/auth" "github.com/oam-dev/kubevela/pkg/controller/core.oam.dev/v1alpha2/application/assemble" "github.com/oam-dev/kubevela/pkg/cue/model/value" + "github.com/oam-dev/kubevela/pkg/cue/packages" "github.com/oam-dev/kubevela/pkg/cue/process" monitorContext "github.com/oam-dev/kubevela/pkg/monitor/context" "github.com/oam-dev/kubevela/pkg/monitor/metrics" @@ -84,7 +85,7 @@ func (h *AppHandler) GenerateApplicationSteps(ctx monitorContext.Context, var tasks []wfTypes.TaskRunner for _, step := range af.WorkflowSteps { - task, err := generateStep(ctx, app, step, taskDiscover, "") + task, err := generateStep(ctx, app, step, taskDiscover, h.r.pd, pCtx, "") if err != nil { return nil, err } @@ -97,9 +98,13 @@ func generateStep(ctx context.Context, app *v1beta1.Application, step v1beta1.WorkflowStep, taskDiscover wfTypes.TaskDiscover, + pd *packages.PackageDiscover, + pCtx process.Context, parentStepName string) (wfTypes.TaskRunner, error) { options := &wfTypes.GeneratorOptions{ - ID: generateStepID(step.Name, app.Status.Workflow, parentStepName), + ID: generateStepID(step.Name, app.Status.Workflow, parentStepName), + PackageDiscover: pd, + ProcessContext: pCtx, } generatorName := step.Type switch { @@ -125,7 +130,7 @@ func generateStep(ctx context.Context, If: subStep.If, Timeout: subStep.Timeout, } - subTask, err := generateStep(ctx, app, workflowStep, taskDiscover, step.Name) + subTask, err := generateStep(ctx, app, workflowStep, taskDiscover, pd, pCtx, step.Name) if err != nil { return nil, err } diff --git a/pkg/controller/core.oam.dev/v1alpha2/application/workflow_test.go b/pkg/controller/core.oam.dev/v1alpha2/application/workflow_test.go index aea5c2937..87f97b7a7 100644 --- a/pkg/controller/core.oam.dev/v1alpha2/application/workflow_test.go +++ b/pkg/controller/core.oam.dev/v1alpha2/application/workflow_test.go @@ -40,7 +40,7 @@ import ( oamcore "github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1" "github.com/oam-dev/kubevela/pkg/oam/testutil" "github.com/oam-dev/kubevela/pkg/oam/util" - "github.com/oam-dev/kubevela/pkg/workflow/tasks/custom" + wfTypes "github.com/oam-dev/kubevela/pkg/workflow/types" ) var _ = Describe("Test Workflow", func() { @@ -315,7 +315,7 @@ var _ = Describe("Test Workflow", func() { appObj.Status.Workflow.Terminated = true appObj.Status.Workflow.Suspend = false appObj.Status.Workflow.Steps[0].Phase = common.WorkflowStepPhaseFailed - appObj.Status.Workflow.Steps[0].Reason = custom.StatusReasonTerminate + appObj.Status.Workflow.Steps[0].Reason = wfTypes.StatusReasonTerminate Expect(k8sClient.Status().Patch(ctx, appObj, client.Merge)).Should(BeNil()) tryReconcile(reconciler, suspendApp.Name, suspendApp.Namespace) diff --git a/pkg/stdlib/op.cue b/pkg/stdlib/op.cue index c325c9088..efef20f0e 100644 --- a/pkg/stdlib/op.cue +++ b/pkg/stdlib/op.cue @@ -15,6 +15,11 @@ import ( message?: string } +#Fail: { + #do: "fail" + message?: string +} + #Apply: kube.#Apply #ApplyInParallel: kube.#ApplyInParallel diff --git a/pkg/workflow/hooks/data_passing.go b/pkg/workflow/hooks/data_passing.go index e23d391c3..9067d2460 100644 --- a/pkg/workflow/hooks/data_passing.go +++ b/pkg/workflow/hooks/data_passing.go @@ -17,7 +17,6 @@ limitations under the License. package hooks import ( - "encoding/json" "strings" "github.com/pkg/errors" @@ -26,6 +25,7 @@ import ( "github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1" "github.com/oam-dev/kubevela/pkg/cue/model/value" wfContext "github.com/oam-dev/kubevela/pkg/workflow/context" + wfTypes "github.com/oam-dev/kubevela/pkg/workflow/types" ) // Input set data to parameter. @@ -43,26 +43,19 @@ func Input(ctx wfContext.Context, paramValue *value.Value, step v1beta1.Workflow } // Output get data from task value. -func Output(ctx wfContext.Context, taskValue *value.Value, step v1beta1.WorkflowStep, phase common.WorkflowStepPhase) error { - if phase == common.WorkflowStepPhaseSucceeded { - if step.Properties != nil { - o := struct { - Name string `json:"name"` - }{} - js, err := common.RawExtensionPointer{RawExtension: step.Properties}.MarshalJSON() - if err != nil { - return err - } - if err := json.Unmarshal(js, &o); err != nil { - return err - } - } - +func Output(ctx wfContext.Context, taskValue *value.Value, step v1beta1.WorkflowStep, status common.StepStatus) error { + if wfTypes.IsStepFinish(status.Phase, status.Reason) { for _, output := range step.Outputs { v, err := taskValue.LookupByScript(output.ValueFrom) if err != nil { return err } + if v.Error() != nil { + v, err = taskValue.MakeValue("null") + if err != nil { + return err + } + } if err := ctx.SetVar(v, output.Name); err != nil { return err } diff --git a/pkg/workflow/hooks/data_passing_test.go b/pkg/workflow/hooks/data_passing_test.go index 813b548aa..8a0f9fd54 100644 --- a/pkg/workflow/hooks/data_passing_test.go +++ b/pkg/workflow/hooks/data_passing_test.go @@ -74,7 +74,9 @@ output: score: 99 ValueFrom: "output.score", Name: "myscore", }}, - }, common.WorkflowStepPhaseSucceeded) + }, common.StepStatus{ + Phase: common.WorkflowStepPhaseSucceeded, + }) r.NoError(err) result, err := wfCtx.GetVar("myscore") r.NoError(err) diff --git a/pkg/workflow/providers/mock/mock.go b/pkg/workflow/providers/mock/mock.go index ad92e636d..18dd54d82 100644 --- a/pkg/workflow/providers/mock/mock.go +++ b/pkg/workflow/providers/mock/mock.go @@ -22,20 +22,26 @@ type Action struct { Message string } -// Suspend ... +// Suspend makes the step suspend func (act *Action) Suspend(message string) { act.Phase = "Suspend" act.Message = message } -// Terminate ... +// Terminate makes the step terminate func (act *Action) Terminate(message string) { act.Phase = "Terminate" act.Message = message } -// Wait ... +// Wait makes the step wait func (act *Action) Wait(message string) { act.Phase = "Wait" act.Message = message } + +// Fail makes the step fail +func (act *Action) Fail(message string) { + act.Phase = "Fail" + act.Message = message +} diff --git a/pkg/workflow/providers/workspace/workspace.go b/pkg/workflow/providers/workspace/workspace.go index c758c1478..1e7588180 100644 --- a/pkg/workflow/providers/workspace/workspace.go +++ b/pkg/workflow/providers/workspace/workspace.go @@ -161,6 +161,16 @@ func (h *provider) Break(ctx wfContext.Context, v *value.Value, act types.Action return nil } +// Fail let the step fail, its status is failed and reason is Action +func (h *provider) Fail(ctx wfContext.Context, v *value.Value, act types.Action) error { + var msg string + if v != nil { + msg, _ = v.GetString("message") + } + act.Fail(msg) + return nil +} + // Install register handler to provider discover. func Install(p providers.Providers) { prd := &provider{} @@ -169,6 +179,7 @@ func Install(p providers.Providers) { "export": prd.Export, "wait": prd.Wait, "break": prd.Break, + "fail": prd.Fail, "var": prd.DoVar, }) } diff --git a/pkg/workflow/providers/workspace/workspace_test.go b/pkg/workflow/providers/workspace/workspace_test.go index e7a096cee..6445a00a2 100644 --- a/pkg/workflow/providers/workspace/workspace_test.go +++ b/pkg/workflow/providers/workspace/workspace_test.go @@ -243,6 +243,25 @@ message: "terminate" assert.Equal(t, act.msg, "terminate") } +func TestProvider_Fail(t *testing.T) { + wfCtx := newWorkflowContextForTest(t) + p := &provider{} + act := &mockAction{} + err := p.Fail(wfCtx, nil, act) + assert.NilError(t, err) + assert.Equal(t, act.terminate, true) + + act = &mockAction{} + v, err := value.NewValue(` +message: "fail" +`, nil, "") + assert.NilError(t, err) + err = p.Fail(wfCtx, v, act) + assert.NilError(t, err) + assert.Equal(t, act.terminate, true) + assert.Equal(t, act.msg, "fail") +} + type mockAction struct { suspend bool terminate bool @@ -259,11 +278,17 @@ func (act *mockAction) Terminate(msg string) { act.terminate = true act.msg = msg } + func (act *mockAction) Wait(msg string) { act.wait = true act.msg = msg } +func (act *mockAction) Fail(msg string) { + act.terminate = true + act.msg = msg +} + func newWorkflowContextForTest(t *testing.T) wfContext.Context { cm := corev1.ConfigMap{} testCaseJson, err := yaml.YAMLToJSON([]byte(testCaseYaml)) diff --git a/pkg/workflow/tasks/custom/task.go b/pkg/workflow/tasks/custom/task.go index 6422888c7..845cccf0b 100644 --- a/pkg/workflow/tasks/custom/task.go +++ b/pkg/workflow/tasks/custom/task.go @@ -24,7 +24,6 @@ import ( "cuelang.org/go/cue" "github.com/pkg/errors" - "k8s.io/apiserver/pkg/util/feature" "github.com/oam-dev/kubevela/apis/core.oam.dev/common" "github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1" @@ -33,7 +32,6 @@ import ( "github.com/oam-dev/kubevela/pkg/cue/model/value" "github.com/oam-dev/kubevela/pkg/cue/packages" "github.com/oam-dev/kubevela/pkg/cue/process" - "github.com/oam-dev/kubevela/pkg/features" monitorContext "github.com/oam-dev/kubevela/pkg/monitor/context" wfContext "github.com/oam-dev/kubevela/pkg/workflow/context" "github.com/oam-dev/kubevela/pkg/workflow/hooks" @@ -41,34 +39,6 @@ import ( wfTypes "github.com/oam-dev/kubevela/pkg/workflow/types" ) -var ( - // MaxWorkflowStepErrorRetryTimes is the max retry times of the failed workflow step. - MaxWorkflowStepErrorRetryTimes = 10 -) - -const ( - // StatusReasonWait is the reason of the workflow progress condition which is Wait. - StatusReasonWait = "Wait" - // StatusReasonSkip is the reason of the workflow progress condition which is Skip. - StatusReasonSkip = "Skip" - // StatusReasonRendering is the reason of the workflow progress condition which is Rendering. - StatusReasonRendering = "Rendering" - // StatusReasonExecute is the reason of the workflow progress condition which is Execute. - StatusReasonExecute = "Execute" - // StatusReasonSuspend is the reason of the workflow progress condition which is Suspend. - StatusReasonSuspend = "Suspend" - // StatusReasonTerminate is the reason of the workflow progress condition which is Terminate. - StatusReasonTerminate = "Terminate" - // StatusReasonParameter is the reason of the workflow progress condition which is ProcessParameter. - StatusReasonParameter = "ProcessParameter" - // StatusReasonOutput is the reason of the workflow progress condition which is Output. - StatusReasonOutput = "Output" - // StatusReasonFailedAfterRetries is the reason of the workflow progress condition which is FailedAfterRetries. - StatusReasonFailedAfterRetries = "FailedAfterRetries" - // StatusReasonTimeout is the reason of the workflow progress condition which is Timeout. - StatusReasonTimeout = "Timeout" -) - // LoadTaskTemplate gets the workflowStep definition from cluster and resolve it. type LoadTaskTemplate func(ctx context.Context, name string) (string, error) @@ -153,7 +123,7 @@ func (t *TaskLoader) makeTaskGenerator(templ string) (wfTypes.TaskGenerator, err tRunner.checkPending = func(ctx wfContext.Context, stepStatus map[string]common.StepStatus) bool { return CheckPending(ctx, wfStep, stepStatus) } - tRunner.run = func(ctx wfContext.Context, options *wfTypes.TaskRunOptions) (common.StepStatus, *wfTypes.Operation, error) { + tRunner.run = func(ctx wfContext.Context, options *wfTypes.TaskRunOptions) (stepStatus common.StepStatus, operations *wfTypes.Operation, rErr error) { if options.GetTracer == nil { options.GetTracer = func(id string, step v1beta1.WorkflowStep) monitorContext.Context { return monitorContext.NewTraceContext(context.Background(), "") @@ -165,11 +135,42 @@ func (t *TaskLoader) makeTaskGenerator(templ string) (wfTypes.TaskGenerator, err tracer.Commit(string(exec.status().Phase)) }() + if t.runOptionsProcess != nil { + t.runOptionsProcess(options) + } + + var taskv *value.Value + var err error + var paramFile string + + defer func() { + if exec.wfStatus.Phase != common.WorkflowStepPhaseSkipped && len(wfStep.Outputs) > 0 { + if taskv == nil { + taskv, err = convertTemplate(ctx, t.pd, strings.Join([]string{templ, paramFile}, "\n"), exec.wfStatus.ID, options.PCtx) + if err != nil { + return + } + } + for _, hook := range options.PostStopHooks { + if err := hook(ctx, taskv, wfStep, exec.status()); err != nil { + exec.err(ctx, false, err, wfTypes.StatusReasonOutput) + stepStatus = exec.status() + operations = exec.operation() + return + } + } + } + }() + for _, hook := range options.PreCheckHooks { - result, err := hook(wfStep) + result, err := hook(wfStep, &wfTypes.PreCheckOptions{ + PackageDiscover: t.pd, + ProcessContext: options.PCtx, + }) if err != nil { tracer.Error(err, "do preCheckHook") - return common.StepStatus{}, nil, errors.WithMessage(err, "do preCheckHook") + exec.Skip(fmt.Sprintf("pre check error: %s", err.Error())) + return exec.status(), exec.operation(), nil } if result.Skip { exec.Skip("") @@ -181,14 +182,6 @@ func (t *TaskLoader) makeTaskGenerator(templ string) (wfTypes.TaskGenerator, err } } - if exec.operation().FailedAfterRetries { - tracer.Info("failed after retries, skip this step") - return exec.status(), exec.operation(), nil - } - - if t.runOptionsProcess != nil { - t.runOptionsProcess(options) - } paramsValue, err := ctx.MakeParameter(params) if err != nil { tracer.Error(err, "make parameter") @@ -203,11 +196,11 @@ func (t *TaskLoader) makeTaskGenerator(templ string) (wfTypes.TaskGenerator, err } if err := paramsValue.Error(); err != nil { - exec.err(ctx, err, StatusReasonParameter) + exec.err(ctx, false, err, wfTypes.StatusReasonParameter) return exec.status(), exec.operation(), nil } - var paramFile = model.ParameterFieldName + ": {}\n" + paramFile = model.ParameterFieldName + ": {}\n" if params != nil { ps, err := paramsValue.String() if err != nil { @@ -216,9 +209,9 @@ func (t *TaskLoader) makeTaskGenerator(templ string) (wfTypes.TaskGenerator, err paramFile = fmt.Sprintf(model.ParameterFieldName+": {%s}\n", ps) } - taskv, err := t.makeValue(ctx, strings.Join([]string{templ, paramFile}, "\n"), exec.wfStatus.ID, options.PCtx) + taskv, err = convertTemplate(ctx, t.pd, strings.Join([]string{templ, paramFile}, "\n"), exec.wfStatus.ID, options.PCtx) if err != nil { - exec.err(ctx, err, StatusReasonRendering) + exec.err(ctx, false, err, wfTypes.StatusReasonRendering) return exec.status(), exec.operation(), nil } @@ -236,40 +229,118 @@ func (t *TaskLoader) makeTaskGenerator(templ string) (wfTypes.TaskGenerator, err } if err := exec.doSteps(ctx, taskv); err != nil { tracer.Error(err, "do steps") - exec.err(ctx, err, StatusReasonExecute) + exec.err(ctx, true, err, wfTypes.StatusReasonExecute) return exec.status(), exec.operation(), nil } - for _, hook := range options.PostStopHooks { - if err := hook(ctx, taskv, wfStep, exec.status().Phase); err != nil { - exec.err(ctx, err, StatusReasonOutput) - return exec.status(), exec.operation(), nil - } - } - return exec.status(), exec.operation(), nil } return tRunner, nil }, nil } -func (t *TaskLoader) makeValue(ctx wfContext.Context, templ string, id string, pCtx process.Context) (*value.Value, error) { +// ValidateIfValue validates the if value +func ValidateIfValue(ctx wfContext.Context, step v1beta1.WorkflowStep, stepStatus map[string]common.StepStatus, options *wfTypes.PreCheckOptions) (bool, error) { + var pd *packages.PackageDiscover + var pCtx process.Context + if options != nil { + pd = options.PackageDiscover + pCtx = options.ProcessContext + } + + template := fmt.Sprintf("if: %s", step.If) + value, err := buildValueForStatus(ctx, step, pd, template, stepStatus, pCtx) + if err != nil { + return false, errors.WithMessage(err, "invalid if value") + } + check, err := value.GetBool("if") + if err != nil { + return false, err + } + return check, nil +} + +func buildValueForStatus(ctx wfContext.Context, step v1beta1.WorkflowStep, pd *packages.PackageDiscover, template string, stepStatus map[string]common.StepStatus, pCtx process.Context) (*value.Value, error) { + contextTempl := getContextTemplate(ctx, "", pCtx) + inputsTempl := getInputsTemplate(ctx, step) + statusTemplate := "\n" + statusMap := make(map[string]interface{}) + for name, ss := range stepStatus { + abbrStatus := struct { + common.StepStatus `json:",inline"` + Failed bool `json:"failed"` + Succeeded bool `json:"succeeded"` + Skipped bool `json:"skipped"` + Timeout bool `json:"timeout"` + FailedAfterRetries bool `json:"failedAfterRetries"` + Terminate bool `json:"terminate"` + }{ + StepStatus: ss, + Failed: ss.Phase == common.WorkflowStepPhaseFailed, + Succeeded: ss.Phase == common.WorkflowStepPhaseSucceeded, + Skipped: ss.Phase == common.WorkflowStepPhaseSkipped, + Timeout: ss.Reason == wfTypes.StatusReasonTimeout, + FailedAfterRetries: ss.Reason == wfTypes.StatusReasonFailedAfterRetries, + Terminate: ss.Reason == wfTypes.StatusReasonTerminate, + } + statusMap[name] = abbrStatus + } + status, err := json.Marshal(statusMap) + if err != nil { + return nil, err + } + statusTemplate += fmt.Sprintf("status: %s\n", status) + statusTemplate += contextTempl + statusTemplate += "\n" + inputsTempl + return value.NewValue(template+"\n"+statusTemplate, pd, statusTemplate) +} + +func convertTemplate(ctx wfContext.Context, pd *packages.PackageDiscover, templ, id string, pCtx process.Context) (*value.Value, error) { + contextTempl := getContextTemplate(ctx, id, pCtx) + return value.NewValue(templ+contextTempl, pd, contextTempl, value.ProcessScript, value.TagFieldOrder) +} + +// MakeValueForContext makes context value +func MakeValueForContext(ctx wfContext.Context, pd *packages.PackageDiscover, id string, pCtx process.Context) (*value.Value, error) { + contextTempl := getContextTemplate(ctx, id, pCtx) + return value.NewValue(contextTempl, pd, contextTempl) +} + +func getContextTemplate(ctx wfContext.Context, id string, pCtx process.Context) string { var contextTempl string meta, _ := ctx.GetVar(wfTypes.ContextKeyMetadata) if meta != nil { ms, err := meta.String() if err != nil { - return nil, err + return "" } contextTempl = fmt.Sprintf("\ncontext: {%s}\ncontext: stepSessionID: \"%s\"", ms, id) } + if pCtx == nil { + return "" + } c, err := pCtx.ExtendedContextFile() if err != nil { - return nil, err + return "" } contextTempl += "\n" + c + return contextTempl +} - return value.NewValue(templ+contextTempl, t.pd, contextTempl, value.ProcessScript, value.TagFieldOrder) +func getInputsTemplate(ctx wfContext.Context, step v1beta1.WorkflowStep) string { + var inputsTempl string + for _, input := range step.Inputs { + inputValue, err := ctx.GetVar(strings.Split(input.From, ".")...) + if err != nil { + continue + } + s, err := inputValue.String() + if err != nil { + continue + } + inputsTempl += fmt.Sprintf("\ninputs: %s: %s", strings.ReplaceAll(input.From, "-", "_"), s) + } + return inputsTempl } type executor struct { @@ -290,7 +361,7 @@ func (exec *executor) Suspend(message string) { exec.suspend = true exec.wfStatus.Phase = common.WorkflowStepPhaseSucceeded exec.wfStatus.Message = message - exec.wfStatus.Reason = StatusReasonSuspend + exec.wfStatus.Reason = wfTypes.StatusReasonSuspend } // Terminate let workflow terminate. @@ -298,33 +369,41 @@ func (exec *executor) Terminate(message string) { exec.terminated = true exec.wfStatus.Phase = common.WorkflowStepPhaseSucceeded exec.wfStatus.Message = message - exec.wfStatus.Reason = StatusReasonTerminate + exec.wfStatus.Reason = wfTypes.StatusReasonTerminate } // Wait let workflow wait. func (exec *executor) Wait(message string) { exec.wait = true exec.wfStatus.Phase = common.WorkflowStepPhaseRunning - exec.wfStatus.Reason = StatusReasonWait + exec.wfStatus.Reason = wfTypes.StatusReasonWait + exec.wfStatus.Message = message +} + +// Fail let the step fail, its status is failed and reason is Action +func (exec *executor) Fail(message string) { + exec.terminated = true + exec.wfStatus.Phase = common.WorkflowStepPhaseFailed + exec.wfStatus.Reason = wfTypes.StatusReasonAction exec.wfStatus.Message = message } func (exec *executor) Skip(message string) { exec.skip = true exec.wfStatus.Phase = common.WorkflowStepPhaseSkipped - exec.wfStatus.Reason = StatusReasonSkip + exec.wfStatus.Reason = wfTypes.StatusReasonSkip exec.wfStatus.Message = message } func (exec *executor) timeout(message string) { exec.terminated = true exec.wfStatus.Phase = common.WorkflowStepPhaseFailed - exec.wfStatus.Reason = StatusReasonTimeout + exec.wfStatus.Reason = wfTypes.StatusReasonTimeout exec.wfStatus.Message = message } -func (exec *executor) err(ctx wfContext.Context, err error, reason string) { - exec.wait = true +func (exec *executor) err(ctx wfContext.Context, wait bool, err error, reason string) { + exec.wait = wait exec.wfStatus.Phase = common.WorkflowStepPhaseFailed exec.wfStatus.Message = err.Error() exec.wfStatus.Reason = reason @@ -333,10 +412,10 @@ func (exec *executor) err(ctx wfContext.Context, err error, reason string) { func (exec *executor) checkErrorTimes(ctx wfContext.Context) { times := ctx.IncreaseCountValueInMemory(wfTypes.ContextPrefixFailedTimes, exec.wfStatus.ID) - if times >= MaxWorkflowStepErrorRetryTimes { + if times >= wfTypes.MaxWorkflowStepErrorRetryTimes { exec.wait = false exec.failedAfterRetries = true - exec.wfStatus.Reason = StatusReasonFailedAfterRetries + exec.wfStatus.Reason = wfTypes.StatusReasonFailedAfterRetries } } @@ -470,38 +549,23 @@ func NewTaskLoader(lt LoadTaskTemplate, pkgDiscover *packages.PackageDiscover, h pd: pkgDiscover, handlers: handlers, runOptionsProcess: func(options *wfTypes.TaskRunOptions) { - options.PreStartHooks = append(options.PreStartHooks, hooks.Input) - options.PostStopHooks = append(options.PostStopHooks, hooks.Output) + if len(options.PreStartHooks) == 0 { + options.PreStartHooks = append(options.PreStartHooks, hooks.Input) + } + if len(options.PostStopHooks) == 0 { + options.PostStopHooks = append(options.PostStopHooks, hooks.Output) + } options.PCtx = pCtx }, logLevel: logLevel, } } -// SkipOptions is the options of skip task runner -type SkipOptions struct { - If string - DependsOnPhase common.WorkflowStepPhase -} - -// SkipTaskRunner will decide whether to skip task runner. -func SkipTaskRunner(options *SkipOptions) bool { - switch options.If { - case "always": - return false - case "": - return options.DependsOnPhase != common.WorkflowStepPhaseSucceeded - default: - // TODO:(fog) support more if cases - return false - } -} - // CheckPending checks whether to pending task run func CheckPending(ctx wfContext.Context, step v1beta1.WorkflowStep, stepStatus map[string]common.StepStatus) bool { for _, depend := range step.DependsOn { if status, ok := stepStatus[depend]; ok { - if !IsStepFinish(status.Phase, status.Reason) { + if !wfTypes.IsStepFinish(status.Phase, status.Reason) { return true } } else { @@ -515,20 +579,3 @@ func CheckPending(ctx wfContext.Context, step v1beta1.WorkflowStep, stepStatus m } return false } - -// IsStepFinish will decide whether step is finish. -func IsStepFinish(phase common.WorkflowStepPhase, reason string) bool { - if feature.DefaultMutableFeatureGate.Enabled(features.EnableSuspendOnFailure) { - return phase == common.WorkflowStepPhaseSucceeded - } - switch phase { - case common.WorkflowStepPhaseFailed: - return reason == StatusReasonTerminate || reason == StatusReasonFailedAfterRetries || reason == StatusReasonTimeout - case common.WorkflowStepPhaseSkipped: - return true - case common.WorkflowStepPhaseSucceeded: - return true - default: - return false - } -} diff --git a/pkg/workflow/tasks/custom/task_test.go b/pkg/workflow/tasks/custom/task_test.go index 3d0d4d90f..0a00d23ad 100644 --- a/pkg/workflow/tasks/custom/task_test.go +++ b/pkg/workflow/tasks/custom/task_test.go @@ -131,24 +131,24 @@ myIP: value: "1.1.1.1" r.NoError(err) if step.Name == "wait" { r.Equal(status.Phase, common.WorkflowStepPhaseRunning) - r.Equal(status.Reason, StatusReasonWait) + r.Equal(status.Reason, types.StatusReasonWait) r.Equal(status.Message, "I am waiting") continue } if step.Name == "terminate" { r.Equal(action.Terminated, true) - r.Equal(status.Reason, StatusReasonTerminate) + r.Equal(status.Reason, types.StatusReasonTerminate) r.Equal(status.Message, "I am terminated") continue } if step.Name == "rendering" { r.Equal(status.Phase, common.WorkflowStepPhaseFailed) - r.Equal(status.Reason, StatusReasonRendering) + r.Equal(status.Reason, types.StatusReasonRendering) continue } if step.Name == "execute" { r.Equal(status.Phase, common.WorkflowStepPhaseFailed) - r.Equal(status.Reason, StatusReasonExecute) + r.Equal(status.Reason, types.StatusReasonExecute) continue } r.Equal(status.Phase, common.WorkflowStepPhaseSucceeded) @@ -212,14 +212,6 @@ close({ ParameterKey: "prefixIP", }}, }, - { - Name: "output", - Type: "ok", - Outputs: common.StepOutputs{{ - Name: "podIP", - ValueFrom: "myIP", - }}, - }, { Name: "output-var-conflict", Type: "ok", @@ -248,16 +240,19 @@ close({ r.NoError(err) status, operation, err := run.Run(wfCtx, &types.TaskRunOptions{}) switch step.Name { + case "input-err": + r.Equal(operation.Waiting, false) + r.Equal(status.Phase, common.WorkflowStepPhaseFailed) case "input": r.Equal(err.Error(), "do preStartHook: get input from [podIP]: var(path=podIP) not exist") - case "output", "output-var-conflict": - r.Equal(status.Reason, StatusReasonOutput) - r.Equal(operation.Waiting, true) + case "output-var-conflict": + r.Equal(status.Reason, types.StatusReasonOutput) + r.Equal(operation.Waiting, false) r.Equal(status.Phase, common.WorkflowStepPhaseFailed) case "failed-after-retries": wfContext.CleanupMemoryStore("app-v1", "default") newCtx := newWorkflowContextForTest(t) - for i := 0; i < MaxWorkflowStepErrorRetryTimes; i++ { + for i := 0; i < types.MaxWorkflowStepErrorRetryTimes; i++ { status, operation, err = run.Run(newCtx, &types.TaskRunOptions{}) r.NoError(err) r.Equal(operation.Waiting, true) @@ -269,7 +264,7 @@ close({ r.Equal(operation.Waiting, false) r.Equal(operation.FailedAfterRetries, true) r.Equal(status.Phase, common.WorkflowStepPhaseFailed) - r.Equal(status.Reason, StatusReasonFailedAfterRetries) + r.Equal(status.Reason, types.StatusReasonFailedAfterRetries) default: r.Equal(operation.Waiting, true) r.Equal(status.Phase, common.WorkflowStepPhaseFailed) @@ -507,14 +502,14 @@ func TestSkip(t *testing.T) { r.NoError(err) status, operations, err := runner.Run(nil, &types.TaskRunOptions{ PreCheckHooks: []types.TaskPreCheckHook{ - func(step v1beta1.WorkflowStep) (*types.PreCheckResult, error) { + func(step v1beta1.WorkflowStep, options *types.PreCheckOptions) (*types.PreCheckResult, error) { return &types.PreCheckResult{Skip: true}, nil }, }, }) r.NoError(err) r.Equal(status.Phase, common.WorkflowStepPhaseSkipped) - r.Equal(status.Reason, StatusReasonSkip) + r.Equal(status.Reason, types.StatusReasonSkip) r.Equal(operations.Skip, true) } @@ -544,14 +539,112 @@ func TestTimeout(t *testing.T) { ctx := newWorkflowContextForTest(t) status, _, err := runner.Run(ctx, &types.TaskRunOptions{ PreCheckHooks: []types.TaskPreCheckHook{ - func(step v1beta1.WorkflowStep) (*types.PreCheckResult, error) { + func(step v1beta1.WorkflowStep, options *types.PreCheckOptions) (*types.PreCheckResult, error) { return &types.PreCheckResult{Timeout: true}, nil }, }, }) r.NoError(err) r.Equal(status.Phase, common.WorkflowStepPhaseFailed) - r.Equal(status.Reason, StatusReasonTimeout) + r.Equal(status.Reason, types.StatusReasonTimeout) +} + +func TestValidateIfValue(t *testing.T) { + ctx := newWorkflowContextForTest(t) + pCtx := process.NewContext(process.ContextData{ + AppName: "app", + CompName: "app", + Namespace: "default", + AppRevisionName: "app-v1", + }) + + testCases := []struct { + name string + step v1beta1.WorkflowStep + status map[string]common.StepStatus + expected bool + expectedErr string + }{ + { + name: "timeout true", + step: v1beta1.WorkflowStep{ + If: "status.step1.timeout", + }, + status: map[string]common.StepStatus{ + "step1": { + Reason: "Timeout", + }, + }, + expected: true, + }, + { + name: "context true", + step: v1beta1.WorkflowStep{ + If: `context.name == "app"`, + }, + expected: true, + }, + { + name: "failed true", + step: v1beta1.WorkflowStep{ + If: `status.step1.phase != "failed"`, + }, + status: map[string]common.StepStatus{ + "step1": { + Phase: common.WorkflowStepPhaseSucceeded, + }, + }, + expected: true, + }, + { + name: "input true", + step: v1beta1.WorkflowStep{ + If: `inputs.test == "yes"`, + Inputs: common.StepInputs{ + { + From: "test", + }, + }, + }, + expected: true, + }, + { + name: "dash in if", + step: v1beta1.WorkflowStep{ + If: "status.step1-test.timeout", + }, + expectedErr: "invalid if value", + expected: false, + }, + { + name: "dash in status", + step: v1beta1.WorkflowStep{ + If: `status["step1-test"].timeout`, + }, + status: map[string]common.StepStatus{ + "step1-test": { + Reason: "Timeout", + }, + }, + expected: true, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + r := require.New(t) + v, err := ValidateIfValue(ctx, tc.step, tc.status, &types.PreCheckOptions{ + ProcessContext: pCtx, + }) + if tc.expectedErr != "" { + r.Contains(err.Error(), tc.expectedErr) + r.Equal(v, false) + return + } + r.NoError(err) + r.Equal(v, tc.expected) + }) + } } func newWorkflowContextForTest(t *testing.T) wfContext.Context { @@ -578,6 +671,8 @@ func newWorkflowContextForTest(t *testing.T) wfContext.Context { r.NoError(err) v, _ := value.NewValue(`name: "app"`, nil, "") r.NoError(wfCtx.SetVar(v, types.ContextKeyMetadata)) + v, _ = value.NewValue(`"yes"`, nil, "") + r.NoError(wfCtx.SetVar(v, "test")) return wfCtx } diff --git a/pkg/workflow/tasks/discover.go b/pkg/workflow/tasks/discover.go index 8c8b5b41d..de1a5e13b 100644 --- a/pkg/workflow/tasks/discover.go +++ b/pkg/workflow/tasks/discover.go @@ -19,9 +19,12 @@ package tasks import ( "context" "encoding/json" + "fmt" + "strings" "time" "github.com/pkg/errors" + "k8s.io/apimachinery/pkg/runtime" "k8s.io/client-go/rest" "sigs.k8s.io/controller-runtime/pkg/client" @@ -74,6 +77,8 @@ func suspend(step v1beta1.WorkflowStep, opt *types.GeneratorOptions) (types.Task tr := &suspendTaskRunner{ id: opt.ID, step: step, + pd: opt.PackageDiscover, + pCtx: opt.ProcessContext, } return tr, nil @@ -86,6 +91,8 @@ func StepGroup(step v1beta1.WorkflowStep, opt *types.GeneratorOptions) (types.Ta name: step.Name, step: step, subTaskRunners: opt.SubTaskRunners, + pd: opt.PackageDiscover, + pCtx: opt.ProcessContext, }, nil } @@ -114,6 +121,8 @@ func NewTaskDiscoverFromRevision(ctx monitorContext.Context, providerHandlers pr type suspendTaskRunner struct { id string step v1beta1.WorkflowStep + pd *packages.PackageDiscover + pCtx process.Context } // Name return suspend step name. @@ -122,47 +131,70 @@ func (tr *suspendTaskRunner) Name() string { } // Run make workflow suspend. -func (tr *suspendTaskRunner) Run(ctx wfContext.Context, options *types.TaskRunOptions) (common.StepStatus, *types.Operation, error) { - stepStatus := common.StepStatus{ +func (tr *suspendTaskRunner) Run(ctx wfContext.Context, options *types.TaskRunOptions) (stepStatus common.StepStatus, operations *types.Operation, rErr error) { + stepStatus = common.StepStatus{ ID: tr.id, Name: tr.step.Name, Type: types.WorkflowStepTypeSuspend, Phase: common.WorkflowStepPhaseRunning, } - operations := &types.Operation{Suspend: true} + operations = &types.Operation{Suspend: true} - if options != nil { - for _, hook := range options.PreCheckHooks { - result, err := hook(tr.step) - if err != nil { - return common.StepStatus{}, nil, errors.WithMessage(err, "do preCheckHook") - } - switch { - case result.Skip: - stepStatus.Phase = common.WorkflowStepPhaseSkipped - stepStatus.Reason = custom.StatusReasonSkip - operations.Suspend = false - operations.Skip = true - case result.Timeout: - stepStatus.Phase = common.WorkflowStepPhaseFailed - stepStatus.Reason = custom.StatusReasonTimeout - operations.Suspend = false - operations.Terminated = true - default: - continue - } - return stepStatus, operations, nil + status := &stepStatus + defer handleOutput(ctx, status, operations, tr.step, options.PostStopHooks, tr.pd, tr.id, tr.pCtx) + + for _, hook := range options.PreCheckHooks { + result, err := hook(tr.step, &types.PreCheckOptions{ + PackageDiscover: tr.pd, + ProcessContext: tr.pCtx, + }) + if err != nil { + stepStatus.Phase = common.WorkflowStepPhaseSkipped + stepStatus.Reason = types.StatusReasonSkip + stepStatus.Message = fmt.Sprintf("pre check error: %s", err.Error()) + operations.Suspend = false + operations.Skip = true + continue } + switch { + case result.Skip: + stepStatus.Phase = common.WorkflowStepPhaseSkipped + stepStatus.Reason = types.StatusReasonSkip + operations.Suspend = false + operations.Skip = true + case result.Timeout: + stepStatus.Phase = common.WorkflowStepPhaseFailed + stepStatus.Reason = types.StatusReasonTimeout + operations.Suspend = false + operations.Terminated = true + default: + continue + } + return stepStatus, operations, nil } + for _, input := range tr.step.Inputs { + if input.ParameterKey == "duration" { + inputValue, err := ctx.GetVar(strings.Split(input.From, ".")...) + if err != nil { + return common.StepStatus{}, nil, errors.WithMessagef(err, "do preStartHook: get input from [%s]", input.From) + } + d, err := inputValue.String() + if err != nil { + return common.StepStatus{}, nil, errors.WithMessagef(err, "do preStartHook: input value from [%s] is not a valid string", input.From) + } + tr.step.Properties = &runtime.RawExtension{Raw: []byte(`{"duration":` + d + `}`)} + } + } d, err := GetSuspendStepDurationWaiting(tr.step) if err != nil { - return stepStatus, operations, err + stepStatus.Message = fmt.Sprintf("invalid suspend duration: %s", err.Error()) + return stepStatus, operations, nil } if d != 0 { e := options.Engine firstExecuteTime := time.Now() - if ss := e.GetStepStatus(tr.step.Name); !ss.FirstExecuteTime.IsZero() { + if ss := e.GetCommonStepStatus(tr.step.Name); !ss.FirstExecuteTime.IsZero() { firstExecuteTime = ss.FirstExecuteTime.Time } if time.Now().After(firstExecuteTime.Add(d)) { @@ -183,6 +215,8 @@ type stepGroupTaskRunner struct { name string step v1beta1.WorkflowStep subTaskRunners []types.TaskRunner + pd *packages.PackageDiscover + pCtx process.Context } // Name return suspend step name. @@ -196,29 +230,40 @@ func (tr *stepGroupTaskRunner) Pending(ctx wfContext.Context, stepStatus map[str } // Run make workflow step group. -func (tr *stepGroupTaskRunner) Run(ctx wfContext.Context, options *types.TaskRunOptions) (common.StepStatus, *types.Operation, error) { - status := common.StepStatus{ +func (tr *stepGroupTaskRunner) Run(ctx wfContext.Context, options *types.TaskRunOptions) (status common.StepStatus, operations *types.Operation, rErr error) { + status = common.StepStatus{ ID: tr.id, Name: tr.name, Type: types.WorkflowStepTypeStepGroup, } + + pStatus := &status + defer handleOutput(ctx, pStatus, operations, tr.step, options.PostStopHooks, tr.pd, tr.id, tr.pCtx) for _, hook := range options.PreCheckHooks { - result, err := hook(tr.step) + result, err := hook(tr.step, &types.PreCheckOptions{ + PackageDiscover: tr.pd, + ProcessContext: options.PCtx, + }) if err != nil { - return common.StepStatus{}, nil, errors.WithMessage(err, "do preCheckHook") + status.Phase = common.WorkflowStepPhaseSkipped + status.Reason = types.StatusReasonSkip + status.Message = fmt.Sprintf("pre check error: %s", err.Error()) + continue } if result.Skip { status.Phase = common.WorkflowStepPhaseSkipped - status.Reason = custom.StatusReasonSkip + status.Reason = types.StatusReasonSkip options.StepStatus[tr.step.Name] = status break } if result.Timeout { status.Phase = common.WorkflowStepPhaseFailed - status.Reason = custom.StatusReasonTimeout + status.Reason = types.StatusReasonTimeout options.StepStatus[tr.step.Name] = status } } + // step-group has no properties so there is no need to fill in the properties with the input values + // skip input handle here e := options.Engine if len(tr.subTaskRunners) > 0 { e.SetParentRunner(tr.name) @@ -233,8 +278,14 @@ func (tr *stepGroupTaskRunner) Run(ctx wfContext.Context, options *types.TaskRun } e.SetParentRunner("") } - stepStatus := e.GetStepStatus(tr.name) + stepStatus := e.GetStepStatus(tr.name) + status, operations = getStepGroupStatus(status, stepStatus, e.GetOperation(), len(tr.subTaskRunners)) + + return status, operations, nil +} + +func getStepGroupStatus(status common.StepStatus, stepStatus common.WorkflowStepStatus, operation *types.Operation, subTaskRunners int) (common.StepStatus, *types.Operation) { subStepCounts := make(map[string]int) for _, subStepsStatus := range stepStatus.SubStepsStatus { subStepCounts[string(subStepsStatus.Phase)]++ @@ -242,10 +293,10 @@ func (tr *stepGroupTaskRunner) Run(ctx wfContext.Context, options *types.TaskRun } switch { case status.Phase == common.WorkflowStepPhaseSkipped: - return status, &types.Operation{Skip: true}, nil - case status.Phase == common.WorkflowStepPhaseFailed && status.Reason == custom.StatusReasonTimeout: - return status, &types.Operation{Terminated: true}, nil - case len(stepStatus.SubStepsStatus) < len(tr.subTaskRunners): + return status, &types.Operation{Skip: true} + case status.Phase == common.WorkflowStepPhaseFailed && status.Reason == types.StatusReasonTimeout: + return status, &types.Operation{Terminated: true} + case len(stepStatus.SubStepsStatus) < subTaskRunners: status.Phase = common.WorkflowStepPhaseRunning case subStepCounts[string(common.WorkflowStepPhaseRunning)] > 0: status.Phase = common.WorkflowStepPhaseRunning @@ -254,20 +305,22 @@ func (tr *stepGroupTaskRunner) Run(ctx wfContext.Context, options *types.TaskRun case subStepCounts[string(common.WorkflowStepPhaseFailed)] > 0: status.Phase = common.WorkflowStepPhaseFailed switch { - case subStepCounts[custom.StatusReasonFailedAfterRetries] > 0: - status.Reason = custom.StatusReasonFailedAfterRetries - case subStepCounts[custom.StatusReasonTimeout] > 0: - status.Reason = custom.StatusReasonTimeout - case subStepCounts[custom.StatusReasonTerminate] > 0: - status.Reason = custom.StatusReasonTerminate + case subStepCounts[types.StatusReasonFailedAfterRetries] > 0: + status.Reason = types.StatusReasonFailedAfterRetries + case subStepCounts[types.StatusReasonTimeout] > 0: + status.Reason = types.StatusReasonTimeout + case subStepCounts[types.StatusReasonAction] > 0: + status.Reason = types.StatusReasonAction + case subStepCounts[types.StatusReasonTerminate] > 0: + status.Reason = types.StatusReasonTerminate } - case subStepCounts[string(common.WorkflowStepPhaseSkipped)] > 0: + case subStepCounts[string(common.WorkflowStepPhaseSkipped)] > 0 && subStepCounts[string(common.WorkflowStepPhaseSkipped)] == subTaskRunners: status.Phase = common.WorkflowStepPhaseSkipped - status.Reason = custom.StatusReasonSkip + status.Reason = types.StatusReasonSkip default: status.Phase = common.WorkflowStepPhaseSucceeded } - return status, e.GetOperation(), nil + return status, operation } // NewViewTaskDiscover will create a client for load task generator. @@ -310,3 +363,31 @@ func GetSuspendStepDurationWaiting(step v1beta1.WorkflowStep) (time.Duration, er return 0, nil } + +func handleOutput(ctx wfContext.Context, stepStatus *common.StepStatus, operations *types.Operation, step v1beta1.WorkflowStep, postStopHooks []types.TaskPostStopHook, pd *packages.PackageDiscover, id string, pCtx process.Context) { + status := *stepStatus + if status.Phase != common.WorkflowStepPhaseSkipped && len(step.Outputs) > 0 { + contextValue, err := custom.MakeValueForContext(ctx, pd, id, pCtx) + if err != nil { + status.Phase = common.WorkflowStepPhaseFailed + if status.Reason == "" { + status.Reason = types.StatusReasonOutput + } + operations.Terminated = true + status.Message = fmt.Sprintf("make context value error: %s", err.Error()) + return + } + + for _, hook := range postStopHooks { + if err := hook(ctx, contextValue, step, status); err != nil { + status.Phase = common.WorkflowStepPhaseFailed + if status.Reason == "" { + status.Reason = types.StatusReasonOutput + } + operations.Terminated = true + status.Message = fmt.Sprintf("output error: %s", err.Error()) + return + } + } + } +} diff --git a/pkg/workflow/tasks/discover_test.go b/pkg/workflow/tasks/discover_test.go index e417dfbce..febcabaea 100644 --- a/pkg/workflow/tasks/discover_test.go +++ b/pkg/workflow/tasks/discover_test.go @@ -102,33 +102,33 @@ func TestSuspendStep(t *testing.T) { // test skip status, operations, err := runner.Run(nil, &types.TaskRunOptions{ PreCheckHooks: []types.TaskPreCheckHook{ - func(step v1beta1.WorkflowStep) (*types.PreCheckResult, error) { + func(step v1beta1.WorkflowStep, options *types.PreCheckOptions) (*types.PreCheckResult, error) { return &types.PreCheckResult{Skip: true}, nil }, }, }) r.NoError(err) r.Equal(status.Phase, common.WorkflowStepPhaseSkipped) - r.Equal(status.Reason, custom.StatusReasonSkip) + r.Equal(status.Reason, types.StatusReasonSkip) r.Equal(operations.Suspend, false) r.Equal(operations.Skip, true) // test timeout status, operations, err = runner.Run(nil, &types.TaskRunOptions{ PreCheckHooks: []types.TaskPreCheckHook{ - func(step v1beta1.WorkflowStep) (*types.PreCheckResult, error) { + func(step v1beta1.WorkflowStep, options *types.PreCheckOptions) (*types.PreCheckResult, error) { return &types.PreCheckResult{Timeout: true}, nil }, }, }) r.NoError(err) r.Equal(status.Phase, common.WorkflowStepPhaseFailed) - r.Equal(status.Reason, custom.StatusReasonTimeout) + r.Equal(status.Reason, types.StatusReasonTimeout) r.Equal(operations.Suspend, false) r.Equal(operations.Terminated, true) // test run - status, act, err := runner.Run(nil, nil) + status, act, err := runner.Run(nil, &types.TaskRunOptions{}) r.NoError(err) r.Equal(act.Suspend, true) r.Equal(status.ID, "124") @@ -149,6 +149,10 @@ func (e *testEngine) GetStepStatus(stepName string) common.WorkflowStepStatus { return e.stepStatus } +func (e *testEngine) GetCommonStepStatus(stepName string) common.StepStatus { + return common.StepStatus{} +} + func (e *testEngine) SetParentRunner(name string) { } @@ -188,7 +192,7 @@ func TestStepGroupStep(t *testing.T) { // test skip status, operations, err := runner.Run(nil, &types.TaskRunOptions{ PreCheckHooks: []types.TaskPreCheckHook{ - func(step v1beta1.WorkflowStep) (*types.PreCheckResult, error) { + func(step v1beta1.WorkflowStep, options *types.PreCheckOptions) (*types.PreCheckResult, error) { return &types.PreCheckResult{Skip: true}, nil }, }, @@ -200,13 +204,13 @@ func TestStepGroupStep(t *testing.T) { }) r.NoError(err) r.Equal(status.Phase, common.WorkflowStepPhaseSkipped) - r.Equal(status.Reason, custom.StatusReasonSkip) + r.Equal(status.Reason, types.StatusReasonSkip) r.Equal(operations.Skip, true) // test timeout status, operations, err = runner.Run(nil, &types.TaskRunOptions{ PreCheckHooks: []types.TaskPreCheckHook{ - func(step v1beta1.WorkflowStep) (*types.PreCheckResult, error) { + func(step v1beta1.WorkflowStep, options *types.PreCheckOptions) (*types.PreCheckResult, error) { return &types.PreCheckResult{Timeout: true}, nil }, }, @@ -218,7 +222,7 @@ func TestStepGroupStep(t *testing.T) { }) r.NoError(err) r.Equal(status.Phase, common.WorkflowStepPhaseFailed) - r.Equal(status.Reason, custom.StatusReasonTimeout) + r.Equal(status.Reason, types.StatusReasonTimeout) r.Equal(operations.Terminated, true) // test run diff --git a/pkg/workflow/types/types.go b/pkg/workflow/types/types.go index a4ce7f4da..ec15d1a65 100644 --- a/pkg/workflow/types/types.go +++ b/pkg/workflow/types/types.go @@ -19,10 +19,14 @@ package types import ( "context" + "k8s.io/apiserver/pkg/util/feature" + "github.com/oam-dev/kubevela/apis/core.oam.dev/common" "github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1" "github.com/oam-dev/kubevela/pkg/cue/model/value" + "github.com/oam-dev/kubevela/pkg/cue/packages" "github.com/oam-dev/kubevela/pkg/cue/process" + "github.com/oam-dev/kubevela/pkg/features" monitorCtx "github.com/oam-dev/kubevela/pkg/monitor/context" wfContext "github.com/oam-dev/kubevela/pkg/workflow/context" ) @@ -43,6 +47,7 @@ type TaskDiscover interface { type Engine interface { Run(taskRunners []TaskRunner, dag bool) error GetStepStatus(stepName string) common.WorkflowStepStatus + GetCommonStepStatus(stepName string) common.StepStatus SetParentRunner(name string) GetOperation() *Operation } @@ -67,14 +72,20 @@ type PreCheckResult struct { Timeout bool } +// PreCheckOptions is the options for pre check. +type PreCheckOptions struct { + PackageDiscover *packages.PackageDiscover + ProcessContext process.Context +} + // TaskPreCheckHook is the hook for pre check. -type TaskPreCheckHook func(step v1beta1.WorkflowStep) (*PreCheckResult, error) +type TaskPreCheckHook func(step v1beta1.WorkflowStep, options *PreCheckOptions) (*PreCheckResult, error) // TaskPreStartHook run before task execution. type TaskPreStartHook func(ctx wfContext.Context, paramValue *value.Value, step v1beta1.WorkflowStep) error // TaskPostStopHook run after task execution. -type TaskPostStopHook func(ctx wfContext.Context, taskValue *value.Value, step v1beta1.WorkflowStep, phase common.WorkflowStepPhase) error +type TaskPostStopHook func(ctx wfContext.Context, taskValue *value.Value, step v1beta1.WorkflowStep, status common.StepStatus) error // Operation is workflow operation object. type Operation struct { @@ -90,10 +101,12 @@ type TaskGenerator func(wfStep v1beta1.WorkflowStep, options *GeneratorOptions) // GeneratorOptions is the options for generate task. type GeneratorOptions struct { - ID string - PrePhase common.WorkflowStepPhase - StepConvertor func(step v1beta1.WorkflowStep) (v1beta1.WorkflowStep, error) - SubTaskRunners []TaskRunner + ID string + PrePhase common.WorkflowStepPhase + StepConvertor func(step v1beta1.WorkflowStep) (v1beta1.WorkflowStep, error) + SubTaskRunners []TaskRunner + PackageDiscover *packages.PackageDiscover + ProcessContext process.Context } // Action is that workflow provider can do. @@ -101,6 +114,7 @@ type Action interface { Suspend(message string) Terminate(message string) Wait(message string) + Fail(message string) } const ( @@ -128,3 +142,54 @@ const ( // WorkflowStepTypeStepGroup type step-group WorkflowStepTypeStepGroup = "step-group" ) + +var ( + // MaxWorkflowStepErrorRetryTimes is the max retry times of the failed workflow step. + MaxWorkflowStepErrorRetryTimes = 10 + // MaxWorkflowWaitBackoffTime is the max time to wait before reconcile wait workflow again + MaxWorkflowWaitBackoffTime = 60 + // MaxWorkflowFailedBackoffTime is the max time to wait before reconcile failed workflow again + MaxWorkflowFailedBackoffTime = 300 +) + +const ( + // StatusReasonWait is the reason of the workflow progress condition which is Wait. + StatusReasonWait = "Wait" + // StatusReasonSkip is the reason of the workflow progress condition which is Skip. + StatusReasonSkip = "Skip" + // StatusReasonRendering is the reason of the workflow progress condition which is Rendering. + StatusReasonRendering = "Rendering" + // StatusReasonExecute is the reason of the workflow progress condition which is Execute. + StatusReasonExecute = "Execute" + // StatusReasonSuspend is the reason of the workflow progress condition which is Suspend. + StatusReasonSuspend = "Suspend" + // StatusReasonTerminate is the reason of the workflow progress condition which is Terminate. + StatusReasonTerminate = "Terminate" + // StatusReasonParameter is the reason of the workflow progress condition which is ProcessParameter. + StatusReasonParameter = "ProcessParameter" + // StatusReasonOutput is the reason of the workflow progress condition which is Output. + StatusReasonOutput = "Output" + // StatusReasonFailedAfterRetries is the reason of the workflow progress condition which is FailedAfterRetries. + StatusReasonFailedAfterRetries = "FailedAfterRetries" + // StatusReasonTimeout is the reason of the workflow progress condition which is Timeout. + StatusReasonTimeout = "Timeout" + // StatusReasonAction is the reason of the workflow progress condition which is Action. + StatusReasonAction = "Action" +) + +// IsStepFinish will decide whether step is finish. +func IsStepFinish(phase common.WorkflowStepPhase, reason string) bool { + if feature.DefaultMutableFeatureGate.Enabled(features.EnableSuspendOnFailure) { + return phase == common.WorkflowStepPhaseSucceeded + } + switch phase { + case common.WorkflowStepPhaseFailed: + return reason != "" && reason != StatusReasonExecute + case common.WorkflowStepPhaseSkipped: + return true + case common.WorkflowStepPhaseSucceeded: + return true + default: + return false + } +} diff --git a/pkg/workflow/workflow.go b/pkg/workflow/workflow.go index 68fff6553..7ba8598b2 100644 --- a/pkg/workflow/workflow.go +++ b/pkg/workflow/workflow.go @@ -41,6 +41,7 @@ import ( "github.com/oam-dev/kubevela/pkg/resourcekeeper" wfContext "github.com/oam-dev/kubevela/pkg/workflow/context" "github.com/oam-dev/kubevela/pkg/workflow/debug" + "github.com/oam-dev/kubevela/pkg/workflow/hooks" "github.com/oam-dev/kubevela/pkg/workflow/recorder" wfTasks "github.com/oam-dev/kubevela/pkg/workflow/tasks" "github.com/oam-dev/kubevela/pkg/workflow/tasks/custom" @@ -52,10 +53,6 @@ var ( DisableRecorder = false // StepStatusCache cache the step status StepStatusCache sync.Map - // MaxWorkflowWaitBackoffTime is the max time to wait before reconcile wait workflow again - MaxWorkflowWaitBackoffTime = 60 - // MaxWorkflowFailedBackoffTime is the max time to wait before reconcile failed workflow again - MaxWorkflowFailedBackoffTime = 300 ) const ( @@ -129,6 +126,13 @@ func (w *workflow) ExecuteSteps(ctx monitorContext.Context, appRev *oamcore.Appl return common.WorkflowStateSucceeded, nil } + if cacheValue, ok := StepStatusCache.Load(cacheKey); ok { + // handle cache resource + if len(wfStatus.Steps) < cacheValue.(int) { + return common.WorkflowStateSkipping, nil + } + } + wfCtx, err := w.makeContext(w.app.Name) if err != nil { ctx.Error(err, "make context") @@ -137,13 +141,6 @@ func (w *workflow) ExecuteSteps(ctx monitorContext.Context, appRev *oamcore.Appl } w.wfCtx = wfCtx - if cacheValue, ok := StepStatusCache.Load(cacheKey); ok { - // handle cache resource - if len(wfStatus.Steps) < cacheValue.(int) { - return common.WorkflowStateSkipping, nil - } - } - e := newEngine(ctx, wfCtx, w, wfStatus) err = e.Run(taskRunners, w.dagMode) @@ -379,8 +376,8 @@ func (w *workflow) allDone(taskRunners []wfTypes.TaskRunner) (bool, bool) { done := false for _, ss := range status.Steps { if ss.Name == t.Name() { - done = custom.IsStepFinish(ss.Phase, ss.Reason) - success = done && (ss.Phase == common.WorkflowStepPhaseSucceeded) + done = wfTypes.IsStepFinish(ss.Phase, ss.Reason) + success = done && (ss.Phase == common.WorkflowStepPhaseSucceeded || ss.Phase == common.WorkflowStepPhaseSkipped) break } } @@ -479,10 +476,10 @@ func (e *engine) getBackoffWaitTime() int { func (e *engine) getMaxBackoffWaitTime() int { for _, step := range e.status.Steps { if step.Phase == common.WorkflowStepPhaseFailed { - return MaxWorkflowFailedBackoffTime + return wfTypes.MaxWorkflowFailedBackoffTime } } - return MaxWorkflowWaitBackoffTime + return wfTypes.MaxWorkflowWaitBackoffTime } func (e *engine) getNextTimeout() int64 { @@ -540,7 +537,7 @@ func (e *engine) runAsDAG(taskRunners []wfTypes.TaskRunner) error { var stepID string if status, ok := e.stepStatus[tRunner.Name()]; ok { stepID = status.ID - finish = custom.IsStepFinish(status.Phase, status.Reason) + finish = wfTypes.IsStepFinish(status.Phase, status.Reason) } if !finish { done = false @@ -606,7 +603,7 @@ func (e *engine) steps(taskRunners []wfTypes.TaskRunner, dag bool) error { wfCtx := e.wfCtx for index, runner := range taskRunners { if status, ok := e.stepStatus[runner.Name()]; ok { - if custom.IsStepFinish(status.Phase, status.Reason) { + if wfTypes.IsStepFinish(status.Phase, status.Reason) { continue } } @@ -622,7 +619,7 @@ func (e *engine) steps(taskRunners []wfTypes.TaskRunner, dag bool) error { e.failedAfterRetries = e.failedAfterRetries || operation.FailedAfterRetries e.waiting = e.waiting || operation.Waiting // for the suspend step with duration, there's no need to increase the backoff time in reconcile when it's still running - if !custom.IsStepFinish(status.Phase, status.Reason) && !isWaitSuspendStep(status) { + if !wfTypes.IsStepFinish(status.Phase, status.Reason) && !isWaitSuspendStep(status) { if err := handleBackoffTimes(wfCtx, status, false); err != nil { return err } @@ -657,31 +654,40 @@ func (e *engine) generateRunOptions(dependsOnPhase common.WorkflowStepPhase) *wf StepStatus: e.stepStatus, Engine: e, PreCheckHooks: []wfTypes.TaskPreCheckHook{ - func(step oamcore.WorkflowStep) (*wfTypes.PreCheckResult, error) { + func(step oamcore.WorkflowStep, options *wfTypes.PreCheckOptions) (*wfTypes.PreCheckResult, error) { if feature.DefaultMutableFeatureGate.Enabled(features.EnableSuspendOnFailure) { return &wfTypes.PreCheckResult{Skip: false}, nil } + if e.parentRunner != "" { + if status, ok := e.stepStatus[e.parentRunner]; ok && status.Phase == common.WorkflowStepPhaseSkipped { + return &wfTypes.PreCheckResult{Skip: true}, nil + } + } switch step.If { case "always": return &wfTypes.PreCheckResult{Skip: false}, nil case "": - return &wfTypes.PreCheckResult{Skip: dependsOnPhase != common.WorkflowStepPhaseSucceeded}, nil + return &wfTypes.PreCheckResult{Skip: isUnsuccessfulStep(dependsOnPhase)}, nil default: - // TODO:(fog) support more if cases - return &wfTypes.PreCheckResult{Skip: false}, nil + ifValue, err := custom.ValidateIfValue(e.wfCtx, step, e.stepStatus, options) + if err != nil { + return &wfTypes.PreCheckResult{Skip: true}, err + } + return &wfTypes.PreCheckResult{Skip: !ifValue}, nil } }, - func(step oamcore.WorkflowStep) (*wfTypes.PreCheckResult, error) { + func(step oamcore.WorkflowStep, options *wfTypes.PreCheckOptions) (*wfTypes.PreCheckResult, error) { status := e.stepStatus[step.Name] if e.parentRunner != "" { - if status, ok := e.stepStatus[e.parentRunner]; ok && status.Phase == common.WorkflowStepPhaseFailed && status.Reason == custom.StatusReasonTimeout { + if status, ok := e.stepStatus[e.parentRunner]; ok && status.Phase == common.WorkflowStepPhaseFailed && status.Reason == wfTypes.StatusReasonTimeout { return &wfTypes.PreCheckResult{Timeout: true}, nil } } if !status.FirstExecuteTime.Time.IsZero() && step.Timeout != "" { duration, err := time.ParseDuration(step.Timeout) if err != nil { - return nil, err + // if the timeout is a invalid duration, return {timeout: false} + return &wfTypes.PreCheckResult{Timeout: false}, err } timeout := status.FirstExecuteTime.Add(duration) e.stepTimeout[step.Name] = timeout @@ -692,6 +698,8 @@ func (e *engine) generateRunOptions(dependsOnPhase common.WorkflowStepPhase) *wf return &wfTypes.PreCheckResult{Timeout: false}, nil }, }, + PreStartHooks: []wfTypes.TaskPreStartHook{hooks.Input}, + PostStopHooks: []wfTypes.TaskPostStopHook{hooks.Output}, } if e.debug { options.Debug = func(step string, v *value.Value) error { @@ -822,11 +830,6 @@ func IsFailedAfterRetry(app *oamcore.Application) bool { } func (e *engine) findDependPhase(taskRunners []wfTypes.TaskRunner, index int, dag bool) common.WorkflowStepPhase { - if e.parentRunner != "" { - if status, ok := e.stepStatus[e.parentRunner]; ok && status.Phase == common.WorkflowStepPhaseSkipped { - return common.WorkflowStepPhaseSkipped - } - } if dag { return e.findDependsOnPhase(taskRunners[index].Name()) } @@ -834,7 +837,7 @@ func (e *engine) findDependPhase(taskRunners []wfTypes.TaskRunner, index int, da return common.WorkflowStepPhaseSucceeded } for i := index - 1; i >= 0; i-- { - if e.stepStatus[taskRunners[i].Name()].Phase != common.WorkflowStepPhaseSucceeded { + if isUnsuccessfulStep(e.stepStatus[taskRunners[i].Name()].Phase) { return e.stepStatus[taskRunners[i].Name()].Phase } } @@ -846,13 +849,17 @@ func (e *engine) findDependsOnPhase(name string) common.WorkflowStepPhase { if e.stepStatus[dependsOn].Phase != common.WorkflowStepPhaseSucceeded { return e.stepStatus[dependsOn].Phase } - if result := e.findDependsOnPhase(dependsOn); result != common.WorkflowStepPhaseSucceeded { + if result := e.findDependsOnPhase(dependsOn); isUnsuccessfulStep(result) { return result } } return common.WorkflowStepPhaseSucceeded } +func isUnsuccessfulStep(phase common.WorkflowStepPhase) bool { + return phase != common.WorkflowStepPhaseSucceeded && phase != common.WorkflowStepPhaseSkipped +} + func isWaitSuspendStep(step common.StepStatus) bool { return step.Type == wfTypes.WorkflowStepTypeSuspend && step.Phase == common.WorkflowStepPhaseRunning } @@ -877,12 +884,12 @@ func handleBackoffTimes(wfCtx wfContext.Context, status common.StepStatus, clear func (e *engine) cleanBackoffTimesForTerminated() { for _, ss := range e.status.Steps { for _, sub := range ss.SubStepsStatus { - if sub.Reason == custom.StatusReasonTerminate { + if sub.Reason == wfTypes.StatusReasonTerminate { e.wfCtx.DeleteValueInMemory(wfTypes.ContextPrefixBackoffTimes, sub.ID) e.wfCtx.DeleteValueInMemory(wfTypes.ContextPrefixBackoffReason, sub.ID) } } - if ss.Reason == custom.StatusReasonTerminate { + if ss.Reason == wfTypes.StatusReasonTerminate { e.wfCtx.DeleteValueInMemory(wfTypes.ContextPrefixBackoffTimes, ss.ID) e.wfCtx.DeleteValueInMemory(wfTypes.ContextPrefixBackoffReason, ss.ID) } @@ -899,6 +906,13 @@ func (e *engine) GetStepStatus(stepName string) common.WorkflowStepStatus { return common.WorkflowStepStatus{} } +func (e *engine) GetCommonStepStatus(stepName string) common.StepStatus { + if status, ok := e.stepStatus[stepName]; ok { + return status + } + return common.StepStatus{} +} + func (e *engine) SetParentRunner(name string) { e.parentRunner = name } diff --git a/pkg/workflow/workflow_test.go b/pkg/workflow/workflow_test.go index 48ee960a2..de015f872 100644 --- a/pkg/workflow/workflow_test.go +++ b/pkg/workflow/workflow_test.go @@ -43,7 +43,6 @@ import ( monitorContext "github.com/oam-dev/kubevela/pkg/monitor/context" wfContext "github.com/oam-dev/kubevela/pkg/workflow/context" "github.com/oam-dev/kubevela/pkg/workflow/tasks" - "github.com/oam-dev/kubevela/pkg/workflow/tasks/custom" wfTypes "github.com/oam-dev/kubevela/pkg/workflow/types" ) @@ -241,6 +240,7 @@ var _ = Describe("Test Workflow", func() { }, { Name: "s2", + If: "status.s1.succeeded", Type: "running", Timeout: "1s", }, @@ -248,6 +248,11 @@ var _ = Describe("Test Workflow", func() { Name: "s3", Type: "success", }, + { + Name: "s4", + If: "status.s2.timeout", + Type: "success", + }, }) ctx := monitorContext.NewTraceContext(context.Background(), "test-app") wf := NewWorkflow(app, k8sClient, common.WorkflowModeStep, false, nil) @@ -282,14 +287,20 @@ var _ = Describe("Test Workflow", func() { Name: "s2", Type: "running", Phase: common.WorkflowStepPhaseFailed, - Reason: custom.StatusReasonTimeout, + Reason: wfTypes.StatusReasonTimeout, }, }, { StepStatus: common.StepStatus{ Name: "s3", Type: "success", Phase: common.WorkflowStepPhaseSkipped, - Reason: custom.StatusReasonSkip, + Reason: wfTypes.StatusReasonSkip, + }, + }, { + StepStatus: common.StepStatus{ + Name: "s4", + Type: "success", + Phase: common.WorkflowStepPhaseSucceeded, }, }, }, @@ -304,6 +315,7 @@ var _ = Describe("Test Workflow", func() { }, { Name: "s2", + If: "status.s1.timeout", Type: "suspend", Timeout: "1s", }, @@ -311,6 +323,16 @@ var _ = Describe("Test Workflow", func() { Name: "s3", Type: "success", }, + { + Name: "s4", + If: "status.s1.succeeded", + Type: "suspend", + Timeout: "1s", + }, + { + Name: "s5", + Type: "success", + }, }) ctx := monitorContext.NewTraceContext(context.Background(), "test-app") wf := NewWorkflow(app, k8sClient, common.WorkflowModeStep, false, nil) @@ -344,15 +366,28 @@ var _ = Describe("Test Workflow", func() { StepStatus: common.StepStatus{ Name: "s2", Type: "suspend", - Phase: common.WorkflowStepPhaseFailed, - Reason: custom.StatusReasonTimeout, + Phase: common.WorkflowStepPhaseSkipped, + Reason: wfTypes.StatusReasonSkip, }, }, { StepStatus: common.StepStatus{ - Name: "s3", + Name: "s3", + Type: "success", + Phase: common.WorkflowStepPhaseSucceeded, + }, + }, { + StepStatus: common.StepStatus{ + Name: "s4", + Type: "suspend", + Phase: common.WorkflowStepPhaseFailed, + Reason: wfTypes.StatusReasonTimeout, + }, + }, { + StepStatus: common.StepStatus{ + Name: "s5", Type: "success", Phase: common.WorkflowStepPhaseSkipped, - Reason: custom.StatusReasonSkip, + Reason: wfTypes.StatusReasonSkip, }, }, }, @@ -422,7 +457,7 @@ var _ = Describe("Test Workflow", func() { Name: "s2", Type: "step-group", Phase: common.WorkflowStepPhaseFailed, - Reason: custom.StatusReasonTimeout, + Reason: wfTypes.StatusReasonTimeout, }, SubStepsStatus: []common.WorkflowSubStepStatus{{ StepStatus: common.StepStatus{ @@ -435,14 +470,14 @@ var _ = Describe("Test Workflow", func() { Name: "s2-sub2", Type: "running", Phase: common.WorkflowStepPhaseFailed, - Reason: custom.StatusReasonTimeout, + Reason: wfTypes.StatusReasonTimeout, }, }, { StepStatus: common.StepStatus{ Name: "s2-suspend", Type: "suspend", Phase: common.WorkflowStepPhaseFailed, - Reason: custom.StatusReasonTimeout, + Reason: wfTypes.StatusReasonTimeout, }, }}, }, { @@ -450,7 +485,7 @@ var _ = Describe("Test Workflow", func() { Name: "s3", Type: "success", Phase: common.WorkflowStepPhaseSkipped, - Reason: custom.StatusReasonSkip, + Reason: wfTypes.StatusReasonSkip, }, }}, })).Should(BeEquivalentTo("")) @@ -516,7 +551,7 @@ var _ = Describe("Test Workflow", func() { Name: "s2", Type: "step-group", Phase: common.WorkflowStepPhaseFailed, - Reason: custom.StatusReasonTimeout, + Reason: wfTypes.StatusReasonTimeout, }, SubStepsStatus: []common.WorkflowSubStepStatus{{ StepStatus: common.StepStatus{ @@ -529,14 +564,14 @@ var _ = Describe("Test Workflow", func() { Name: "s2-sub2", Type: "running", Phase: common.WorkflowStepPhaseFailed, - Reason: custom.StatusReasonTimeout, + Reason: wfTypes.StatusReasonTimeout, }, }, { StepStatus: common.StepStatus{ Name: "s2-suspend", Type: "suspend", Phase: common.WorkflowStepPhaseFailed, - Reason: custom.StatusReasonTimeout, + Reason: wfTypes.StatusReasonTimeout, }, }}, }, { @@ -544,7 +579,7 @@ var _ = Describe("Test Workflow", func() { Name: "s3", Type: "success", Phase: common.WorkflowStepPhaseSkipped, - Reason: custom.StatusReasonSkip, + Reason: wfTypes.StatusReasonSkip, }, }}, })).Should(BeEquivalentTo("")) @@ -596,28 +631,28 @@ var _ = Describe("Test Workflow", func() { Name: "s1", Type: "failed-after-retries", Phase: common.WorkflowStepPhaseFailed, - Reason: custom.StatusReasonFailedAfterRetries, + Reason: wfTypes.StatusReasonFailedAfterRetries, }, }, { StepStatus: common.StepStatus{ Name: "s2", Type: "step-group", Phase: common.WorkflowStepPhaseSkipped, - Reason: custom.StatusReasonSkip, + Reason: wfTypes.StatusReasonSkip, }, SubStepsStatus: []common.WorkflowSubStepStatus{{ StepStatus: common.StepStatus{ Name: "s2-sub1", Type: "success", Phase: common.WorkflowStepPhaseSkipped, - Reason: custom.StatusReasonSkip, + Reason: wfTypes.StatusReasonSkip, }, }, { StepStatus: common.StepStatus{ Name: "s2-sub2", Type: "failed", Phase: common.WorkflowStepPhaseSkipped, - Reason: custom.StatusReasonSkip, + Reason: wfTypes.StatusReasonSkip, }, }}, }, { @@ -625,7 +660,7 @@ var _ = Describe("Test Workflow", func() { Name: "s3", Type: "success", Phase: common.WorkflowStepPhaseSkipped, - Reason: custom.StatusReasonSkip, + Reason: wfTypes.StatusReasonSkip, }, }}, })).Should(BeEquivalentTo("")) @@ -680,14 +715,14 @@ var _ = Describe("Test Workflow", func() { Name: "s1", Type: "failed-after-retries", Phase: common.WorkflowStepPhaseFailed, - Reason: custom.StatusReasonFailedAfterRetries, + Reason: wfTypes.StatusReasonFailedAfterRetries, }, }, { StepStatus: common.StepStatus{ Name: "s2", Type: "step-group", Phase: common.WorkflowStepPhaseFailed, - Reason: custom.StatusReasonFailedAfterRetries, + Reason: wfTypes.StatusReasonFailedAfterRetries, }, SubStepsStatus: []common.WorkflowSubStepStatus{{ StepStatus: common.StepStatus{ @@ -700,7 +735,7 @@ var _ = Describe("Test Workflow", func() { Name: "s2-sub2", Type: "failed-after-retries", Phase: common.WorkflowStepPhaseFailed, - Reason: custom.StatusReasonFailedAfterRetries, + Reason: wfTypes.StatusReasonFailedAfterRetries, }, }}, }, { @@ -708,7 +743,7 @@ var _ = Describe("Test Workflow", func() { Name: "s3", Type: "success", Phase: common.WorkflowStepPhaseSkipped, - Reason: custom.StatusReasonSkip, + Reason: wfTypes.StatusReasonSkip, }, }}, })).Should(BeEquivalentTo("")) @@ -886,7 +921,7 @@ var _ = Describe("Test Workflow", func() { Name: "s2", Type: "failed-after-retries", Phase: common.WorkflowStepPhaseFailed, - Reason: custom.StatusReasonFailedAfterRetries, + Reason: wfTypes.StatusReasonFailedAfterRetries, }, }}, })).Should(BeEquivalentTo("")) @@ -935,7 +970,7 @@ var _ = Describe("Test Workflow", func() { Name: "s2", Type: "failed-after-retries", Phase: common.WorkflowStepPhaseFailed, - Reason: custom.StatusReasonFailedAfterRetries, + Reason: wfTypes.StatusReasonFailedAfterRetries, }, }, { StepStatus: common.StepStatus{ @@ -1003,7 +1038,7 @@ var _ = Describe("Test Workflow", func() { Name: "s2", Type: "failed-after-retries", Phase: common.WorkflowStepPhaseFailed, - Reason: custom.StatusReasonFailedAfterRetries, + Reason: wfTypes.StatusReasonFailedAfterRetries, }, }, { StepStatus: common.StepStatus{ @@ -1016,7 +1051,7 @@ var _ = Describe("Test Workflow", func() { Name: "s4", Type: "success", Phase: common.WorkflowStepPhaseSkipped, - Reason: custom.StatusReasonSkip, + Reason: wfTypes.StatusReasonSkip, }, }, { StepStatus: common.StepStatus{ @@ -1089,7 +1124,7 @@ var _ = Describe("Test Workflow", func() { Name: "s2", Type: "failed-after-retries", Phase: common.WorkflowStepPhaseFailed, - Reason: custom.StatusReasonFailedAfterRetries, + Reason: wfTypes.StatusReasonFailedAfterRetries, }, }, { StepStatus: common.StepStatus{ @@ -1102,7 +1137,7 @@ var _ = Describe("Test Workflow", func() { Name: "s4", Type: "success", Phase: common.WorkflowStepPhaseSkipped, - Reason: custom.StatusReasonSkip, + Reason: wfTypes.StatusReasonSkip, }, }, { StepStatus: common.StepStatus{ @@ -1115,12 +1150,283 @@ var _ = Describe("Test Workflow", func() { Name: "s6", Type: "success", Phase: common.WorkflowStepPhaseSkipped, - Reason: custom.StatusReasonSkip, + Reason: wfTypes.StatusReasonSkip, }, }}, })).Should(BeEquivalentTo("")) }) + It("Workflow test if expressions", func() { + By("Test if expressions in StepByStep mode") + app, runners := makeTestCase([]oamcore.WorkflowStep{ + { + Name: "s1", + Type: "success", + }, + { + Name: "s2", + If: "status.s1.failed", + Type: "success", + }, + { + Name: "s3", + If: "status.s1.succeeded", + Type: "success", + Outputs: common.StepOutputs{ + { + Name: "test", + ValueFrom: "context.name", + }, + }, + }, + { + Name: "s4", + Inputs: common.StepInputs{ + { + From: "test", + ParameterKey: "", + }, + }, + If: `inputs.test == "app"`, + Type: "success", + }, + }) + ctx := monitorContext.NewTraceContext(context.Background(), "test-app") + wf := NewWorkflow(app, k8sClient, common.WorkflowModeStep, false, nil) + state, err := wf.ExecuteSteps(ctx, revision, runners) + Expect(err).ToNot(HaveOccurred()) + Expect(state).Should(BeEquivalentTo(common.WorkflowStateInitializing)) + + state, err = wf.ExecuteSteps(ctx, revision, runners) + Expect(err).ToNot(HaveOccurred()) + Expect(state).Should(BeEquivalentTo(common.WorkflowStateSucceeded)) + workflowStatus := app.Status.Workflow + Expect(workflowStatus.ContextBackend.Name).Should(BeEquivalentTo("workflow-" + app.Name + "-context")) + workflowStatus.ContextBackend = nil + cleanStepTimeStamp(workflowStatus) + Expect(cmp.Diff(*workflowStatus, common.WorkflowStatus{ + AppRevision: workflowStatus.AppRevision, + Mode: common.WorkflowModeStep, + Message: "Succeeded", + Suspend: false, + Steps: []common.WorkflowStepStatus{{ + StepStatus: common.StepStatus{ + Name: "s1", + Type: "success", + Phase: common.WorkflowStepPhaseSucceeded, + }, + }, { + StepStatus: common.StepStatus{ + Name: "s2", + Type: "success", + Phase: common.WorkflowStepPhaseSkipped, + Reason: wfTypes.StatusReasonSkip, + }, + }, { + StepStatus: common.StepStatus{ + Name: "s3", + Type: "success", + Phase: common.WorkflowStepPhaseSucceeded, + }, + }, { + StepStatus: common.StepStatus{ + Name: "s4", + Type: "success", + Phase: common.WorkflowStepPhaseSucceeded, + }, + }}, + })).Should(BeEquivalentTo("")) + + By("Test if expressions in DAG mode") + app, runners = makeTestCase([]oamcore.WorkflowStep{ + { + Name: "s1", + Type: "success", + }, + { + Name: "s2", + If: "status.s1.failed", + Type: "success", + }, + { + Name: "s3", + DependsOn: []string{"s2"}, + If: "status.s1.succeeded", + Type: "success", + Outputs: common.StepOutputs{ + { + Name: "test", + ValueFrom: "context.name", + }, + }, + }, + { + Name: "s4", + DependsOn: []string{"s3"}, + Inputs: common.StepInputs{ + { + From: "test", + ParameterKey: "", + }, + }, + If: `inputs.test == "app"`, + Type: "success", + }, + }) + ctx = monitorContext.NewTraceContext(context.Background(), "test-app") + wf = NewWorkflow(app, k8sClient, common.WorkflowModeDAG, false, nil) + state, err = wf.ExecuteSteps(ctx, revision, runners) + Expect(err).ToNot(HaveOccurred()) + Expect(state).Should(BeEquivalentTo(common.WorkflowStateInitializing)) + + state, err = wf.ExecuteSteps(ctx, revision, runners) + Expect(err).ToNot(HaveOccurred()) + Expect(state).Should(BeEquivalentTo(common.WorkflowStateSucceeded)) + workflowStatus = app.Status.Workflow + Expect(workflowStatus.ContextBackend.Name).Should(BeEquivalentTo("workflow-" + app.Name + "-context")) + workflowStatus.ContextBackend = nil + cleanStepTimeStamp(workflowStatus) + Expect(cmp.Diff(*workflowStatus, common.WorkflowStatus{ + AppRevision: workflowStatus.AppRevision, + Mode: common.WorkflowModeDAG, + Message: "Succeeded", + Steps: []common.WorkflowStepStatus{{ + StepStatus: common.StepStatus{ + Name: "s1", + Type: "success", + Phase: common.WorkflowStepPhaseSucceeded, + }, + }, { + StepStatus: common.StepStatus{ + Name: "s2", + Type: "success", + Phase: common.WorkflowStepPhaseSkipped, + Reason: wfTypes.StatusReasonSkip, + }, + }, { + StepStatus: common.StepStatus{ + Name: "s3", + Type: "success", + Phase: common.WorkflowStepPhaseSucceeded, + }, + }, { + StepStatus: common.StepStatus{ + Name: "s4", + Type: "success", + Phase: common.WorkflowStepPhaseSucceeded, + }, + }}, + })).Should(BeEquivalentTo("")) + }) + + It("Workflow test if expressions with sub steps", func() { + By("Test if expressions with step group") + app, runners := makeTestCase([]oamcore.WorkflowStep{ + { + Name: "s1", + Type: "success", + }, + { + Name: "s2", + If: "status.s1.timeout", + Type: "step-group", + SubSteps: []common.WorkflowSubStep{ + { + Name: "s2_sub1", + If: "always", + Type: "success", + }, + { + Name: "s2_sub2", + Type: "failed-after-retries", + }, + }, + }, + { + Name: "s3", + If: "status.s1.succeeded", + Type: "step-group", + SubSteps: []common.WorkflowSubStep{ + { + Name: "s3_sub1", + If: "status.s2_sub1.skipped", + Type: "success", + }, + { + Name: "s3_sub2", + Type: "failed-after-retries", + }, + }, + }, + }) + wf := NewWorkflow(app, k8sClient, common.WorkflowModeStep, false, nil) + ctx := monitorContext.NewTraceContext(context.Background(), "test-app") + state, err := wf.ExecuteSteps(ctx, revision, runners) + Expect(err).ToNot(HaveOccurred()) + Expect(state).Should(BeEquivalentTo(common.WorkflowStateInitializing)) + state, err = wf.ExecuteSteps(ctx, revision, runners) + Expect(err).ToNot(HaveOccurred()) + Expect(state).Should(BeEquivalentTo(common.WorkflowStateTerminated)) + app.Status.Workflow.ContextBackend = nil + cleanStepTimeStamp(app.Status.Workflow) + Expect(cmp.Diff(*app.Status.Workflow, common.WorkflowStatus{ + AppRevision: app.Status.Workflow.AppRevision, + Mode: common.WorkflowModeStep, + Terminated: true, + Message: string(MessageTerminatedFailedAfterRetries), + Steps: []common.WorkflowStepStatus{{ + StepStatus: common.StepStatus{ + Name: "s1", + Type: "success", + Phase: common.WorkflowStepPhaseSucceeded, + }, + }, { + StepStatus: common.StepStatus{ + Name: "s2", + Type: "step-group", + Phase: common.WorkflowStepPhaseSkipped, + Reason: wfTypes.StatusReasonSkip, + }, + SubStepsStatus: []common.WorkflowSubStepStatus{{ + StepStatus: common.StepStatus{ + Name: "s2_sub1", + Type: "success", + Phase: common.WorkflowStepPhaseSkipped, + Reason: wfTypes.StatusReasonSkip, + }, + }, { + StepStatus: common.StepStatus{ + Name: "s2_sub2", + Type: "failed-after-retries", + Phase: common.WorkflowStepPhaseSkipped, + Reason: wfTypes.StatusReasonSkip, + }, + }}, + }, { + StepStatus: common.StepStatus{ + Name: "s3", + Type: "step-group", + Phase: common.WorkflowStepPhaseFailed, + Reason: wfTypes.StatusReasonFailedAfterRetries, + }, + SubStepsStatus: []common.WorkflowSubStepStatus{{ + StepStatus: common.StepStatus{ + Name: "s3_sub1", + Type: "success", + Phase: common.WorkflowStepPhaseSucceeded, + }, + }, { + StepStatus: common.StepStatus{ + Name: "s3_sub2", + Type: "failed-after-retries", + Phase: common.WorkflowStepPhaseFailed, + Reason: wfTypes.StatusReasonFailedAfterRetries, + }, + }}, + }}, + })).Should(BeEquivalentTo("")) + }) + It("Test failed after retries with sub steps", func() { By("Test failed-after-retries with step group in StepByStep mode") defer featuregatetesting.SetFeatureGateDuringTest(&testing.T{}, utilfeature.DefaultFeatureGate, features.EnableSuspendOnFailure, true)() @@ -1174,7 +1480,7 @@ var _ = Describe("Test Workflow", func() { Name: "s2", Type: "step-group", Phase: common.WorkflowStepPhaseFailed, - Reason: custom.StatusReasonFailedAfterRetries, + Reason: wfTypes.StatusReasonFailedAfterRetries, }, SubStepsStatus: []common.WorkflowSubStepStatus{{ StepStatus: common.StepStatus{ @@ -1187,7 +1493,7 @@ var _ = Describe("Test Workflow", func() { Name: "s2-sub2", Type: "failed-after-retries", Phase: common.WorkflowStepPhaseFailed, - Reason: custom.StatusReasonFailedAfterRetries, + Reason: wfTypes.StatusReasonFailedAfterRetries, }, }}, }}, @@ -1234,7 +1540,7 @@ var _ = Describe("Test Workflow", func() { _, err = wf.ExecuteSteps(ctx, revision, runners) Expect(err).ToNot(HaveOccurred()) interval = e.getBackoffWaitTime() - Expect(interval).Should(BeEquivalentTo(MaxWorkflowWaitBackoffTime)) + Expect(interval).Should(BeEquivalentTo(wfTypes.MaxWorkflowWaitBackoffTime)) By("Test get backoff time after clean") wfContext.CleanupMemoryStore(app.Name, app.Namespace) @@ -1866,6 +2172,13 @@ func makeRunner(step oamcore.WorkflowStep, subTaskRunners []wfTypes.TaskRunner) } case "success": run = func(ctx wfContext.Context, options *wfTypes.TaskRunOptions) (common.StepStatus, *wfTypes.Operation, error) { + v, err := value.NewValue(`"app"`, nil, "") + if err != nil { + return common.StepStatus{}, nil, err + } + if err := ctx.SetVar(v, "test"); err != nil { + return common.StepStatus{}, nil, err + } return common.StepStatus{ Name: step.Name, Type: "success", @@ -1886,7 +2199,7 @@ func makeRunner(step oamcore.WorkflowStep, subTaskRunners []wfTypes.TaskRunner) Name: step.Name, Type: "failed-after-retries", Phase: common.WorkflowStepPhaseFailed, - Reason: custom.StatusReasonFailedAfterRetries, + Reason: wfTypes.StatusReasonFailedAfterRetries, }, &wfTypes.Operation{ FailedAfterRetries: true, }, nil @@ -1976,7 +2289,9 @@ func (tr *testTaskRunner) Name() string { func (tr *testTaskRunner) Run(ctx wfContext.Context, options *wfTypes.TaskRunOptions) (common.StepStatus, *wfTypes.Operation, error) { if tr.step.Type != "step-group" && options != nil { for _, hook := range options.PreCheckHooks { - result, err := hook(tr.step) + result, err := hook(tr.step, &wfTypes.PreCheckOptions{ + ProcessContext: options.PCtx, + }) if err != nil { return common.StepStatus{}, nil, errors.WithMessage(err, "do preCheckHook") } @@ -1985,7 +2300,7 @@ func (tr *testTaskRunner) Run(ctx wfContext.Context, options *wfTypes.TaskRunOpt Name: tr.step.Name, Type: tr.step.Type, Phase: common.WorkflowStepPhaseSkipped, - Reason: custom.StatusReasonSkip, + Reason: wfTypes.StatusReasonSkip, }, &wfTypes.Operation{Skip: true}, nil } if result.Timeout { @@ -1993,7 +2308,7 @@ func (tr *testTaskRunner) Run(ctx wfContext.Context, options *wfTypes.TaskRunOpt Name: tr.step.Name, Type: tr.step.Type, Phase: common.WorkflowStepPhaseFailed, - Reason: custom.StatusReasonTimeout, + Reason: wfTypes.StatusReasonTimeout, }, &wfTypes.Operation{Terminated: true}, nil } } diff --git a/references/cli/workflow_test.go b/references/cli/workflow_test.go index 222247e86..147564f4a 100644 --- a/references/cli/workflow_test.go +++ b/references/cli/workflow_test.go @@ -31,7 +31,7 @@ import ( "github.com/oam-dev/kubevela/apis/core.oam.dev/common" "github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1" cmdutil "github.com/oam-dev/kubevela/pkg/utils/util" - "github.com/oam-dev/kubevela/pkg/workflow/tasks/custom" + wfTypes "github.com/oam-dev/kubevela/pkg/workflow/types" ) var workflowSpec = v1beta1.ApplicationSpec{ @@ -386,14 +386,13 @@ func TestWorkflowTerminate(t *testing.T) { r.Equal(true, wf.Status.Workflow.Terminated) for _, step := range wf.Status.Workflow.Steps { if step.Phase != common.WorkflowStepPhaseSucceeded { - fmt.Println("======", step.Name) r.Equal(step.Phase, common.WorkflowStepPhaseFailed) - r.Equal(step.Reason, custom.StatusReasonTerminate) + r.Equal(step.Reason, wfTypes.StatusReasonTerminate) } for _, sub := range step.SubStepsStatus { if sub.Phase != common.WorkflowStepPhaseSucceeded { r.Equal(sub.Phase, common.WorkflowStepPhaseFailed) - r.Equal(sub.Reason, custom.StatusReasonTerminate) + r.Equal(sub.Reason, wfTypes.StatusReasonTerminate) } } }