Feat(vela): add vela workflow terminate and restart command (#2131)

* Feat(vela): add vela workflow terminate and restart command

* Fix(restart): fix restart command

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

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

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

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
 	}
 

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 <application-name>",
+		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 <application-name>",
+		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
+}

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)
+		})
+	}
+}