diff --git a/apis/core.oam.dev/common/types.go b/apis/core.oam.dev/common/types.go index d3253fd59..da025e2c4 100644 --- a/apis/core.oam.dev/common/types.go +++ b/apis/core.oam.dev/common/types.go @@ -169,6 +169,8 @@ const ( ApplicationRunningWorkflow ApplicationPhase = "runningWorkflow" // ApplicationWorkflowSuspending means the app's workflow is suspending ApplicationWorkflowSuspending ApplicationPhase = "workflowSuspending" + // ApplicationWorkflowTerminated means the app's workflow is terminated + ApplicationWorkflowTerminated ApplicationPhase = "workflowTerminated" // ApplicationRunning means the app finished rendering and applied result to the cluster ApplicationRunning ApplicationPhase = "running" // ApplicationHealthChecking means the app finished rendering and applied result to the cluster, but still unhealthy diff --git a/pkg/controller/core.oam.dev/v1alpha2/application/application_controller.go b/pkg/controller/core.oam.dev/v1alpha2/application/application_controller.go index 89f1a217e..b4e8d23d6 100644 --- a/pkg/controller/core.oam.dev/v1alpha2/application/application_controller.go +++ b/pkg/controller/core.oam.dev/v1alpha2/application/application_controller.go @@ -79,6 +79,7 @@ type Reconciler struct { // +kubebuilder:rbac:groups=core.oam.dev,resources=applications/status,verbs=get;update;patch // Reconcile process app event +// nolint:gocyclo func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { ctx, cancel := common2.NewReconcileContext(ctx) defer cancel() @@ -194,17 +195,27 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu return reconcile.Result{RequeueAfter: WorkflowReconcileWaitTime}, r.patchStatus(ctx, app) } - if wfStatus := app.Status.Workflow; wfStatus != nil && !wfStatus.Terminated { - ref, err := handler.DispatchAndGC(ctx) - if err != nil { - klog.ErrorS(err, "Failed to gc after workflow", - "application", klog.KObj(app)) - r.Recorder.Event(app, event.Warning(velatypes.ReasonFailedGC, err)) - return r.endWithNegativeCondition(ctx, app, condition.ErrorCondition("GCAfterWorkflow", err)) + wfStatus := app.Status.Workflow + if wfStatus != nil { + if wfStatus.Terminated && app.Status.Phase == common.ApplicationWorkflowTerminated { + if err := r.patchStatus(ctx, app); err != nil { + return r.endWithNegativeCondition(ctx, app, condition.ReconcileError(err)) + } + return ctrl.Result{}, nil + } + + if !wfStatus.Terminated { + ref, err := handler.DispatchAndGC(ctx) + if err != nil { + klog.ErrorS(err, "Failed to gc after workflow", + "application", klog.KObj(app)) + r.Recorder.Event(app, event.Warning(velatypes.ReasonFailedGC, err)) + return r.endWithNegativeCondition(ctx, app, condition.ErrorCondition("GCAfterWorkflow", err)) + } + wfStatus.Terminated = true + app.Status.ResourceTracker = ref + return r.endWithNegativeCondition(ctx, app, condition.ReadyCondition("GCAfterWorkflow")) } - wfStatus.Terminated = true - app.Status.ResourceTracker = ref - return r.endWithNegativeCondition(ctx, app, condition.ReadyCondition("GCAfterWorkflow")) } // if inplace is false and rolloutPlan is nil, it means the user will use an outer AppRollout object to rollout the application 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 929b149fe..4c316cc05 100644 --- a/pkg/controller/core.oam.dev/v1alpha2/application/workflow_test.go +++ b/pkg/controller/core.oam.dev/v1alpha2/application/workflow_test.go @@ -227,6 +227,54 @@ var _ = Describe("Test Workflow", func() { Expect(appObj.Status.Workflow.StepIndex).Should(BeEquivalentTo(1)) }) + It("test workflow terminate a suspend workflow", func() { + suspendApp := appWithWorkflow.DeepCopy() + suspendApp.Name = "test-terminate-suspend-app" + suspendApp.Spec.Workflow.Steps = []oamcore.WorkflowStep{ + { + Name: "suspend", + Type: "suspend", + Properties: runtime.RawExtension{Raw: []byte(`{}`)}, + }, + { + Name: "suspend", + Type: "suspend", + Properties: runtime.RawExtension{Raw: []byte(`{}`)}, + }} + Expect(k8sClient.Create(ctx, suspendApp)).Should(BeNil()) + + // first try to add finalizer + tryReconcile(reconciler, suspendApp.Name, suspendApp.Namespace) + tryReconcile(reconciler, suspendApp.Name, suspendApp.Namespace) + + appObj := &oamcore.Application{} + Expect(k8sClient.Get(ctx, client.ObjectKey{ + Name: suspendApp.Name, + Namespace: suspendApp.Namespace, + }, appObj)).Should(BeNil()) + + Expect(appObj.Status.Workflow.Suspend).Should(BeTrue()) + Expect(appObj.Status.Phase).Should(BeEquivalentTo(common.ApplicationWorkflowSuspending)) + + // terminate + appObj.Status.Workflow.Terminated = true + Expect(k8sClient.Status().Patch(ctx, appObj, client.Merge)).Should(BeNil()) + + tryReconcile(reconciler, suspendApp.Name, suspendApp.Namespace) + tryReconcile(reconciler, suspendApp.Name, suspendApp.Namespace) + + appObj = &oamcore.Application{} + Expect(k8sClient.Get(ctx, client.ObjectKey{ + Name: suspendApp.Name, + Namespace: suspendApp.Namespace, + }, appObj)).Should(BeNil()) + + Expect(appObj.Status.Workflow.Suspend).Should(BeTrue()) + Expect(appObj.Status.Workflow.Terminated).Should(BeTrue()) + Expect(appObj.Status.Workflow.StepIndex).Should(BeEquivalentTo(1)) + Expect(appObj.Status.Phase).Should(BeEquivalentTo(common.ApplicationWorkflowTerminated)) + }) + }) func triggerWorkflowStepToSucceed(obj *unstructured.Unstructured) { diff --git a/pkg/workflow/workflow.go b/pkg/workflow/workflow.go index e6c83e065..c0bd8fc57 100644 --- a/pkg/workflow/workflow.go +++ b/pkg/workflow/workflow.go @@ -65,6 +65,9 @@ func (w *workflow) ExecuteSteps(ctx context.Context, rev string, taskRunners []w if wfStatus.Terminated { done = true + if len(taskRunners) > wfStatus.StepIndex { + w.app.Status.Phase = common.ApplicationWorkflowTerminated + } return } diff --git a/references/cli/workflow.go b/references/cli/workflow.go index 0a4c95010..e7bdc82bf 100644 --- a/references/cli/workflow.go +++ b/references/cli/workflow.go @@ -43,6 +43,8 @@ func NewWorkflowCommand(c common.Args, ioStreams cmdutil.IOStreams) *cobra.Comma cmd.AddCommand( NewWorkflowSuspendCommand(c, ioStreams), NewWorkflowResumeCommand(c, ioStreams), + NewWorkflowTerminateCommand(c, ioStreams), + NewWorkflowRestartCommand(c, ioStreams), ) return cmd } @@ -111,6 +113,9 @@ func NewWorkflowResumeCommand(c common.Args, ioStream cmdutil.IOStreams) *cobra. if app.Status.Workflow == nil { return fmt.Errorf("the workflow in application is not running") } + if app.Status.Workflow.Terminated { + return fmt.Errorf("can not resume a terminated workflow") + } if !app.Status.Workflow.Suspend { _, err := ioStream.Out.Write([]byte("the workflow is not suspending\n")) if err != nil { @@ -132,6 +137,84 @@ func NewWorkflowResumeCommand(c common.Args, ioStream cmdutil.IOStreams) *cobra. } } +// NewWorkflowTerminateCommand create workflow terminate command +func NewWorkflowTerminateCommand(c common.Args, ioStream cmdutil.IOStreams) *cobra.Command { + return &cobra.Command{ + Use: "terminate", + Short: "Terminate an application workflow", + Long: "Terminate an application workflow in cluster", + Example: "vela workflow terminate ", + RunE: func(cmd *cobra.Command, args []string) error { + if len(args) < 1 { + return fmt.Errorf("must specify application name") + } + env, err := GetEnv(cmd) + if err != nil { + return err + } + app, err := appfile.LoadApplication(env.Namespace, args[0], c) + if err != nil { + return err + } + if app.Spec.Workflow == nil { + return fmt.Errorf("the application must have workflow") + } + if app.Status.Workflow == nil { + return fmt.Errorf("the workflow in application is not running") + } + kubecli, err := c.GetClient() + if err != nil { + return err + } + + err = terminateWorkflow(kubecli, app) + if err != nil { + return err + } + return nil + }, + } +} + +// NewWorkflowRestartCommand create workflow restart command +func NewWorkflowRestartCommand(c common.Args, ioStream cmdutil.IOStreams) *cobra.Command { + return &cobra.Command{ + Use: "restart", + Short: "Restart an application workflow", + Long: "Restart an application workflow in cluster", + Example: "vela workflow restart ", + RunE: func(cmd *cobra.Command, args []string) error { + if len(args) < 1 { + return fmt.Errorf("must specify application name") + } + env, err := GetEnv(cmd) + if err != nil { + return err + } + app, err := appfile.LoadApplication(env.Namespace, args[0], c) + if err != nil { + return err + } + if app.Spec.Workflow == nil { + return fmt.Errorf("the application must have workflow") + } + if app.Status.Workflow == nil { + return fmt.Errorf("the workflow in application is not running") + } + kubecli, err := c.GetClient() + if err != nil { + return err + } + + err = restartWorkflow(kubecli, app) + if err != nil { + return err + } + return nil + }, + } +} + func suspendWorkflow(kubecli client.Client, app *v1beta1.Application) error { // set the workflow suspend to true app.Status.Workflow.Suspend = true @@ -155,3 +238,27 @@ func resumeWorkflow(kubecli client.Client, app *v1beta1.Application) error { fmt.Printf("Successfully resume workflow: %s\n", app.Name) return nil } + +func terminateWorkflow(kubecli client.Client, app *v1beta1.Application) error { + // set the workflow terminated to true + app.Status.Workflow.Terminated = true + + if err := kubecli.Status().Patch(context.TODO(), app, client.Merge); err != nil { + return err + } + + fmt.Printf("Successfully terminate workflow: %s\n", app.Name) + return nil +} + +func restartWorkflow(kubecli client.Client, app *v1beta1.Application) error { + // reset the workflow status to restart the workflow + app.Status.Workflow = nil + + if err := kubecli.Status().Update(context.TODO(), app); err != nil { + return err + } + + fmt.Printf("Successfully restart workflow: %s\n", app.Name) + return nil +} diff --git a/references/cli/workflow_test.go b/references/cli/workflow_test.go index 60fa34715..09be7aad8 100644 --- a/references/cli/workflow_test.go +++ b/references/cli/workflow_test.go @@ -32,6 +32,21 @@ import ( cmdutil "github.com/oam-dev/kubevela/pkg/utils/util" ) +var workflowSpec = v1beta1.ApplicationSpec{ + Components: []common.ApplicationComponent{{ + Name: "test-component", + Type: "worker", + Properties: runtime.RawExtension{Raw: []byte(`{"cmd":["sleep","1000"],"image":"busybox"}`)}, + }}, + Workflow: &v1beta1.Workflow{ + Steps: []v1beta1.WorkflowStep{{ + Name: "test-wf1", + Type: "foowf", + Properties: runtime.RawExtension{Raw: []byte(`{"namespace":"default"}`)}, + }}, + }, +} + func TestWorkflowSuspend(t *testing.T) { c := initArgs() ioStream := cmdutil.IOStreams{In: os.Stdin, Out: os.Stdout, ErrOut: os.Stderr} @@ -53,26 +68,24 @@ func TestWorkflowSuspend(t *testing.T) { }, expectedErr: fmt.Errorf("the application must have workflow"), }, + "workflow not running": { + app: &v1beta1.Application{ + ObjectMeta: metav1.ObjectMeta{ + Name: "workflow-not-running", + Namespace: "default", + }, + Spec: workflowSpec, + Status: common.AppStatus{}, + }, + expectedErr: fmt.Errorf("the workflow in application is not running"), + }, "suspend successfully": { app: &v1beta1.Application{ ObjectMeta: metav1.ObjectMeta{ Name: "workflow", Namespace: "default", }, - Spec: v1beta1.ApplicationSpec{ - Components: []common.ApplicationComponent{{ - Name: "test-component", - Type: "worker", - Properties: runtime.RawExtension{Raw: []byte(`{"cmd":["sleep","1000"],"image":"busybox"}`)}, - }}, - Workflow: &v1beta1.Workflow{ - Steps: []v1beta1.WorkflowStep{{ - Name: "test-wf1", - Type: "foowf", - Properties: runtime.RawExtension{Raw: []byte(`{"namespace":"default"}`)}, - }}, - }, - }, + Spec: workflowSpec, Status: common.AppStatus{ Workflow: &common.WorkflowStatus{ Suspend: false, @@ -139,20 +152,7 @@ func TestWorkflowResume(t *testing.T) { Name: "workflow-not-suspended", Namespace: "default", }, - Spec: v1beta1.ApplicationSpec{ - Components: []common.ApplicationComponent{{ - Name: "test-component", - Type: "worker", - Properties: runtime.RawExtension{Raw: []byte(`{"cmd":["sleep","1000"],"image":"busybox"}`)}, - }}, - Workflow: &v1beta1.Workflow{ - Steps: []v1beta1.WorkflowStep{{ - Name: "test-wf1", - Type: "foowf", - Properties: runtime.RawExtension{Raw: []byte(`{"namespace":"default"}`)}, - }}, - }, - }, + Spec: workflowSpec, Status: common.AppStatus{ Workflow: &common.WorkflowStatus{ Suspend: false, @@ -160,26 +160,39 @@ func TestWorkflowResume(t *testing.T) { }, }, }, + "workflow not running": { + app: &v1beta1.Application{ + ObjectMeta: metav1.ObjectMeta{ + Name: "workflow-not-running", + Namespace: "default", + }, + Spec: workflowSpec, + Status: common.AppStatus{}, + }, + expectedErr: fmt.Errorf("the workflow in application is not running"), + }, + "workflow terminated": { + app: &v1beta1.Application{ + ObjectMeta: metav1.ObjectMeta{ + Name: "workflow-terminated", + Namespace: "default", + }, + Spec: workflowSpec, + Status: common.AppStatus{ + Workflow: &common.WorkflowStatus{ + Terminated: true, + }, + }, + }, + expectedErr: fmt.Errorf("can not resume a terminated workflow"), + }, "resume successfully": { app: &v1beta1.Application{ ObjectMeta: metav1.ObjectMeta{ Name: "workflow", Namespace: "default", }, - Spec: v1beta1.ApplicationSpec{ - Components: []common.ApplicationComponent{{ - Name: "test-component", - Type: "worker", - Properties: runtime.RawExtension{Raw: []byte(`{"cmd":["sleep","1000"],"image":"busybox"}`)}, - }}, - Workflow: &v1beta1.Workflow{ - Steps: []v1beta1.WorkflowStep{{ - Name: "test-wf1", - Type: "foowf", - Properties: runtime.RawExtension{Raw: []byte(`{"namespace":"default"}`)}, - }}, - }, - }, + Spec: workflowSpec, Status: common.AppStatus{ Workflow: &common.WorkflowStatus{ Suspend: true, @@ -218,3 +231,160 @@ func TestWorkflowResume(t *testing.T) { }) } } + +func TestWorkflowTerminate(t *testing.T) { + c := initArgs() + ioStream := cmdutil.IOStreams{In: os.Stdin, Out: os.Stdout, ErrOut: os.Stderr} + ctx := context.TODO() + + testCases := map[string]struct { + app *v1beta1.Application + expectedErr error + }{ + "no app name specified": { + expectedErr: fmt.Errorf("must specify application name"), + }, + "no workflow in app": { + app: &v1beta1.Application{ + ObjectMeta: metav1.ObjectMeta{ + Name: "no-workflow", + Namespace: "default", + }, + }, + expectedErr: fmt.Errorf("the application must have workflow"), + }, + "workflow not running": { + app: &v1beta1.Application{ + ObjectMeta: metav1.ObjectMeta{ + Name: "workflow-not-running", + Namespace: "default", + }, + Spec: workflowSpec, + Status: common.AppStatus{}, + }, + expectedErr: fmt.Errorf("the workflow in application is not running"), + }, + "terminate successfully": { + app: &v1beta1.Application{ + ObjectMeta: metav1.ObjectMeta{ + Name: "workflow", + Namespace: "default", + }, + Spec: workflowSpec, + Status: common.AppStatus{ + Workflow: &common.WorkflowStatus{ + Terminated: false, + }, + }, + }, + }, + } + + for name, tc := range testCases { + t.Run(name, func(t *testing.T) { + r := require.New(t) + cmd := NewWorkflowTerminateCommand(c, ioStream) + initCommand(cmd) + + if tc.app != nil { + err := c.Client.Create(ctx, tc.app) + r.NoError(err) + + cmd.SetArgs([]string{tc.app.Name}) + } + err := cmd.Execute() + if tc.expectedErr != nil { + r.Equal(tc.expectedErr, err) + return + } + r.NoError(err) + + wf := &v1beta1.Application{} + err = c.Client.Get(ctx, types.NamespacedName{ + Namespace: tc.app.Namespace, + Name: tc.app.Name, + }, wf) + r.NoError(err) + r.Equal(true, wf.Status.Workflow.Terminated) + }) + } +} + +func TestWorkflowRestart(t *testing.T) { + c := initArgs() + ioStream := cmdutil.IOStreams{In: os.Stdin, Out: os.Stdout, ErrOut: os.Stderr} + ctx := context.TODO() + + testCases := map[string]struct { + app *v1beta1.Application + expectedErr error + }{ + "no app name specified": { + expectedErr: fmt.Errorf("must specify application name"), + }, + "no workflow in app": { + app: &v1beta1.Application{ + ObjectMeta: metav1.ObjectMeta{ + Name: "no-workflow", + Namespace: "default", + }, + }, + expectedErr: fmt.Errorf("the application must have workflow"), + }, + "workflow not running": { + app: &v1beta1.Application{ + ObjectMeta: metav1.ObjectMeta{ + Name: "workflow-not-running", + Namespace: "default", + }, + Spec: workflowSpec, + Status: common.AppStatus{}, + }, + expectedErr: fmt.Errorf("the workflow in application is not running"), + }, + "restart successfully": { + app: &v1beta1.Application{ + ObjectMeta: metav1.ObjectMeta{ + Name: "workflow", + Namespace: "default", + }, + Spec: workflowSpec, + Status: common.AppStatus{ + Workflow: &common.WorkflowStatus{ + Terminated: true, + }, + }, + }, + }, + } + + for name, tc := range testCases { + t.Run(name, func(t *testing.T) { + r := require.New(t) + cmd := NewWorkflowRestartCommand(c, ioStream) + initCommand(cmd) + + if tc.app != nil { + err := c.Client.Create(ctx, tc.app) + r.NoError(err) + + cmd.SetArgs([]string{tc.app.Name}) + } + err := cmd.Execute() + if tc.expectedErr != nil { + r.Equal(tc.expectedErr, err) + return + } + r.NoError(err) + + wf := &v1beta1.Application{} + err = c.Client.Get(ctx, types.NamespacedName{ + Namespace: tc.app.Namespace, + Name: tc.app.Name, + }, wf) + r.NoError(err) + var nilStatus *common.WorkflowStatus = nil + r.Equal(nilStatus, wf.Status.Workflow) + }) + } +}