Fix: stores workflow status in revison if it is restarted (#5604) (#5673)

Signed-off-by: FogDong <dongtianxin.tx@alibaba-inc.com>

pkg/controller/core.oam.dev/v1alpha2/application/application_controller.go

@@ -193,7 +193,9 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
 	app.Status.SetConditions(condition.ReadyCondition(common.PolicyCondition.String()))
 	r.Recorder.Event(app, event.Normal(velatypes.ReasonPolicyGenerated, velatypes.MessagePolicyGenerated))
 
-	workflowInstance, runners, err := handler.GenerateApplicationSteps(logCtx, app, appParser, appFile, handler.currentAppRev)
+	handler.CheckWorkflowRestart(logCtx, app)
+
+	workflowInstance, runners, err := handler.GenerateApplicationSteps(logCtx, app, appParser, appFile)
 	if err != nil {
 		logCtx.Error(err, "[handle workflow]")
 		r.Recorder.Event(app, event.Warning(velatypes.ReasonFailedWorkflow, err))
@@ -449,15 +451,12 @@ func (r *Reconciler) doWorkflowFinish(logCtx monitorContext.Context, app *v1beta
 	wfContext.CleanupMemoryStore(app.Name, app.Namespace)
 	t := time.Since(app.Status.Workflow.StartTime.Time).Seconds()
 	metrics.WorkflowFinishedTimeHistogram.WithLabelValues(string(state)).Observe(t)
-	switch state {
-	case workflowv1alpha1.WorkflowStateSucceeded:
+	if state == workflowv1alpha1.WorkflowStateSucceeded {
 		app.Status.SetConditions(condition.ReadyCondition(common.WorkflowCondition.String()))
 		r.Recorder.Event(app, event.Normal(velatypes.ReasonApplied, velatypes.MessageWorkflowFinished))
-		handler.UpdateApplicationRevisionStatus(logCtx, handler.currentAppRev, true, app.Status.Workflow)
-		logCtx.Info("Application manifests has applied by workflow successfully")
-	default:
-		handler.UpdateApplicationRevisionStatus(logCtx, handler.latestAppRev, false, app.Status.Workflow)
 	}
+	handler.UpdateApplicationRevisionStatus(logCtx, handler.currentAppRev, app.Status.Workflow)
+	logCtx.Info("Application manifests has applied by workflow successfully")
 }
 
 func hasHealthCheckPolicy(policies []*appfile.Workload) bool {

pkg/controller/core.oam.dev/v1alpha2/application/application_controller_test.go

@@ -719,6 +719,76 @@ var _ = Describe("Test Application Controller", func() {
 		}, appRevision)).Should(BeNil())
 	})
 
+	It("revision should be updated if the workflow is restarted", func() {
+
+		ns := &corev1.Namespace{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "vela-test-app-restart-revision",
+			},
+		}
+		Expect(k8sClient.Create(ctx, ns)).Should(BeNil())
+
+		app := &v1beta1.Application{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "vela-test-app-restart-revision",
+				Namespace: "vela-test-app-restart-revision",
+			},
+			Spec: v1beta1.ApplicationSpec{
+				Components: []common.ApplicationComponent{},
+				Workflow: &v1beta1.Workflow{
+					Steps: []workflowv1alpha1.WorkflowStep{
+						{
+							WorkflowStepBase: workflowv1alpha1.WorkflowStepBase{
+								Name: "suspend",
+								Type: "suspend",
+							},
+						},
+					},
+				},
+			},
+		}
+
+		Expect(k8sClient.Create(ctx, app.DeepCopy())).Should(BeNil())
+
+		appKey := client.ObjectKey{
+			Name:      app.Name,
+			Namespace: app.Namespace,
+		}
+		testutil.ReconcileOnceAfterFinalizer(reconciler, reconcile.Request{NamespacedName: appKey})
+		By("Check Application Created with the correct revision")
+		curApp := &v1beta1.Application{}
+		Expect(k8sClient.Get(ctx, appKey, curApp)).Should(BeNil())
+		Expect(curApp.Status.Phase).Should(Equal(common.ApplicationWorkflowSuspending))
+		Expect(curApp.Status.LatestRevision).ShouldNot(BeNil())
+		Expect(curApp.Status.LatestRevision.Revision).Should(BeEquivalentTo(1))
+
+		appRevision := &v1beta1.ApplicationRevision{}
+		Expect(k8sClient.Get(ctx, client.ObjectKey{
+			Namespace: app.Namespace,
+			Name:      curApp.Status.LatestRevision.Name,
+		}, appRevision)).Should(BeNil())
+		Expect(appRevision.Status.Workflow).Should(BeNil())
+
+		// update the app
+		curApp.Spec.Workflow.Steps[0].DependsOn = []string{"invalid"}
+		Expect(k8sClient.Update(ctx, curApp)).Should(BeNil())
+		Expect(k8sClient.Get(ctx, appKey, curApp)).Should(BeNil())
+		testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey})
+
+		Expect(k8sClient.Get(ctx, client.ObjectKey{
+			Namespace: app.Namespace,
+			Name:      curApp.Status.LatestRevision.Name,
+		}, appRevision)).Should(BeNil())
+		Expect(appRevision.Status.Workflow).ShouldNot(BeNil())
+		Expect(appRevision.Status.Workflow.Finished).Should(BeTrue())
+		Expect(appRevision.Status.Workflow.Terminated).Should(BeTrue())
+		Expect(appRevision.Status.Workflow.EndTime.IsZero()).ShouldNot(BeTrue())
+		Expect(appRevision.Status.Workflow.Phase).Should(Equal(workflowv1alpha1.WorkflowStateSuspending))
+
+		By("Delete Application, clean the resource")
+		Expect(k8sClient.Delete(ctx, app)).Should(BeNil())
+	})
+
 	It("revision should exist in created workload render by context.appRevision", func() {
 
 		ns := &corev1.Namespace{

pkg/controller/core.oam.dev/v1alpha2/application/generator.go

@@ -85,9 +85,9 @@ var (
 func (h *AppHandler) GenerateApplicationSteps(ctx monitorContext.Context,
 	app *v1beta1.Application,
 	appParser *appfile.Parser,
-	af *appfile.Appfile,
-	appRev *v1beta1.ApplicationRevision) (*wfTypes.WorkflowInstance, []wfTypes.TaskRunner, error) {
+	af *appfile.Appfile) (*wfTypes.WorkflowInstance, []wfTypes.TaskRunner, error) {
 
+	appRev := h.currentAppRev
 	appLabels := map[string]string{
 		oam.LabelAppName:      app.Name,
 		oam.LabelAppNamespace: app.Namespace,
@@ -118,7 +118,7 @@ func (h *AppHandler) GenerateApplicationSteps(ctx monitorContext.Context,
 	})
 	query.Install(handlerProviders, h.r.Client, nil)
 
-	instance := generateWorkflowInstance(af, app, appRev.Name)
+	instance := generateWorkflowInstance(af, app)
 	executor.InitializeWorkflowInstance(instance)
 	runners, err := generator.GenerateRunners(ctx, instance, wfTypes.StepGeneratorOptions{
 		Providers:       handlerProviders,
@@ -143,7 +143,7 @@ func (h *AppHandler) GenerateApplicationSteps(ctx monitorContext.Context,
 	return instance, runners, nil
 }
 
-// needRestart check if application workflow need restart and return the desired
+// CheckWorkflowRestart check if application workflow need restart and return the desired
 // rev to be set in status
 // 1. If workflow status is empty, it means no previous running record, the
 // workflow will restart (cold start)
@@ -152,8 +152,8 @@ func (h *AppHandler) GenerateApplicationSteps(ctx monitorContext.Context,
 // 3. If workflow status is not empty, the desired rev will be the
 // ApplicationRevision name. For backward compatibility, the legacy style
 // <rev>:<hash> will be recognized and reduced into <rev>
-func needRestart(app *v1beta1.Application, revName string) (string, bool) {
-	desiredRev, currentRev := revName, ""
+func (h *AppHandler) CheckWorkflowRestart(ctx monitorContext.Context, app *v1beta1.Application) {
+	desiredRev, currentRev := h.currentAppRev.Name, ""
 	if app.Status.Workflow != nil {
 		currentRev = app.Status.Workflow.AppRevision
 	}
@@ -166,10 +166,40 @@ func needRestart(app *v1beta1.Application, revName string) (string, bool) {
 			currentRev = currentRev[:idx]
 		}
 	}
-	return desiredRev, currentRev == "" || desiredRev != currentRev
+	if currentRev != "" && desiredRev == currentRev {
+		return
+	}
+	// record in revision
+	if h.latestAppRev != nil && h.latestAppRev.Status.Workflow == nil && app.Status.Workflow != nil {
+		app.Status.Workflow.Terminated = true
+		app.Status.Workflow.Finished = true
+		if app.Status.Workflow.EndTime.IsZero() {
+			app.Status.Workflow.EndTime = metav1.Now()
+		}
+		h.UpdateApplicationRevisionStatus(ctx, h.latestAppRev, app.Status.Workflow)
+	}
+
+	// clean recorded resources info.
+	app.Status.Services = nil
+	app.Status.AppliedResources = nil
+
+	// clean conditions after render
+	var reservedConditions []condition.Condition
+	for i, cond := range app.Status.Conditions {
+		condTpy, err := common.ParseApplicationConditionType(string(cond.Type))
+		if err == nil {
+			if condTpy <= common.RenderCondition {
+				reservedConditions = append(reservedConditions, app.Status.Conditions[i])
+			}
+		}
+	}
+	app.Status.Conditions = reservedConditions
+	app.Status.Workflow = &common.WorkflowStatus{
+		AppRevision: desiredRev,
+	}
 }
 
-func generateWorkflowInstance(af *appfile.Appfile, app *v1beta1.Application, appRev string) *wfTypes.WorkflowInstance {
+func generateWorkflowInstance(af *appfile.Appfile, app *v1beta1.Application) *wfTypes.WorkflowInstance {
 	instance := &wfTypes.WorkflowInstance{
 		WorkflowMeta: wfTypes.WorkflowMeta{
 			Name:        af.Name,
@@ -191,27 +221,6 @@ func generateWorkflowInstance(af *appfile.Appfile, app *v1beta1.Application, app
 		Steps: af.WorkflowSteps,
 		Mode:  af.WorkflowMode,
 	}
-	if desiredRev, nr := needRestart(app, appRev); nr {
-		// clean recorded resources info.
-		app.Status.Services = nil
-		app.Status.AppliedResources = nil
-
-		// clean conditions after render
-		var reservedConditions []condition.Condition
-		for i, cond := range app.Status.Conditions {
-			condTpy, err := common.ParseApplicationConditionType(string(cond.Type))
-			if err == nil {
-				if condTpy <= common.RenderCondition {
-					reservedConditions = append(reservedConditions, app.Status.Conditions[i])
-				}
-			}
-		}
-		app.Status.Conditions = reservedConditions
-		app.Status.Workflow = &common.WorkflowStatus{
-			AppRevision: desiredRev,
-		}
-		return instance
-	}
 	status := app.Status.Workflow
 	instance.Status = workflowv1alpha1.WorkflowRunStatus{
 		Mode:           *af.WorkflowMode,

pkg/controller/core.oam.dev/v1alpha2/application/generator_test.go

@@ -112,13 +112,15 @@ var _ = Describe("Test Application workflow generator", func() {
 		Expect(err).Should(BeNil())
 		_, err = af.GeneratePolicyManifests(context.Background())
 		Expect(err).Should(BeNil())
-		appRev := &oamcore.ApplicationRevision{}
 
 		handler, err := NewAppHandler(ctx, reconciler, app, appParser)
 		Expect(err).Should(Succeed())
 
 		logCtx := monitorContext.NewTraceContext(ctx, "")
-		_, taskRunner, err := handler.GenerateApplicationSteps(logCtx, app, appParser, af, appRev)
+		handler.currentAppRev = &oamcore.ApplicationRevision{}
+		handler.CheckWorkflowRestart(logCtx, app)
+
+		_, taskRunner, err := handler.GenerateApplicationSteps(logCtx, app, appParser, af)
 		Expect(err).To(BeNil())
 		Expect(len(taskRunner)).Should(BeEquivalentTo(2))
 		Expect(taskRunner[0].Name()).Should(BeEquivalentTo("myweb1"))
@@ -154,13 +156,14 @@ var _ = Describe("Test Application workflow generator", func() {
 		Expect(err).Should(BeNil())
 		_, err = af.GeneratePolicyManifests(context.Background())
 		Expect(err).Should(BeNil())
-		appRev := &oamcore.ApplicationRevision{}
 
 		handler, err := NewAppHandler(ctx, reconciler, app, appParser)
 		Expect(err).Should(Succeed())
 
 		logCtx := monitorContext.NewTraceContext(ctx, "")
-		_, taskRunner, err := handler.GenerateApplicationSteps(logCtx, app, appParser, af, appRev)
+		handler.currentAppRev = &oamcore.ApplicationRevision{}
+		handler.CheckWorkflowRestart(logCtx, app)
+		_, taskRunner, err := handler.GenerateApplicationSteps(logCtx, app, appParser, af)
 		Expect(err).To(BeNil())
 		Expect(len(taskRunner)).Should(BeEquivalentTo(2))
 		Expect(taskRunner[0].Name()).Should(BeEquivalentTo("myweb1"))
@@ -276,13 +279,14 @@ var _ = Describe("Test Application workflow generator", func() {
 		}
 		af, err := appParser.GenerateAppFile(ctx, app)
 		Expect(err).Should(BeNil())
-		appRev := &oamcore.ApplicationRevision{}
 
 		handler, err := NewAppHandler(ctx, reconciler, app, appParser)
 		Expect(err).Should(Succeed())
 
 		logCtx := monitorContext.NewTraceContext(ctx, "")
-		_, taskRunner, err := handler.GenerateApplicationSteps(logCtx, app, appParser, af, appRev)
+		handler.currentAppRev = &oamcore.ApplicationRevision{}
+		handler.CheckWorkflowRestart(logCtx, app)
+		_, taskRunner, err := handler.GenerateApplicationSteps(logCtx, app, appParser, af)
 		Expect(err).To(BeNil())
 		Expect(len(taskRunner)).Should(BeEquivalentTo(2))
 		Expect(taskRunner[0].Name()).Should(BeEquivalentTo("myweb1"))
@@ -317,13 +321,14 @@ var _ = Describe("Test Application workflow generator", func() {
 		}
 		af, err := appParser.GenerateAppFile(ctx, app)
 		Expect(err).Should(BeNil())
-		appRev := &oamcore.ApplicationRevision{}
 
 		handler, err := NewAppHandler(ctx, reconciler, app, appParser)
 		Expect(err).Should(Succeed())
 
 		logCtx := monitorContext.NewTraceContext(ctx, "")
-		_, _, err = handler.GenerateApplicationSteps(logCtx, app, appParser, af, appRev)
+		handler.currentAppRev = &oamcore.ApplicationRevision{}
+		handler.CheckWorkflowRestart(logCtx, app)
+		_, _, err = handler.GenerateApplicationSteps(logCtx, app, appParser, af)
 		Expect(err).NotTo(BeNil())
 	})
 
@@ -355,13 +360,14 @@ var _ = Describe("Test Application workflow generator", func() {
 		}
 		af, err := appParser.GenerateAppFile(ctx, app)
 		Expect(err).Should(BeNil())
-		appRev := &oamcore.ApplicationRevision{}
 
 		handler, err := NewAppHandler(ctx, reconciler, app, appParser)
 		Expect(err).Should(Succeed())
 
 		logCtx := monitorContext.NewTraceContext(ctx, "")
-		_, _, err = handler.GenerateApplicationSteps(logCtx, app, appParser, af, appRev)
+		handler.currentAppRev = &oamcore.ApplicationRevision{}
+		handler.CheckWorkflowRestart(logCtx, app)
+		_, _, err = handler.GenerateApplicationSteps(logCtx, app, appParser, af)
 		Expect(err).NotTo(BeNil())
 	})
 })

pkg/controller/core.oam.dev/v1alpha2/application/revision.go

@@ -1019,11 +1019,11 @@ func (h historiesByComponentRevision) Less(i, j int) bool {
 }
 
 // UpdateApplicationRevisionStatus update application revision status
-func (h *AppHandler) UpdateApplicationRevisionStatus(ctx context.Context, appRev *v1beta1.ApplicationRevision, succeed bool, wfStatus *common.WorkflowStatus) {
+func (h *AppHandler) UpdateApplicationRevisionStatus(ctx context.Context, appRev *v1beta1.ApplicationRevision, wfStatus *common.WorkflowStatus) {
 	if appRev == nil || DisableAllApplicationRevision {
 		return
 	}
-	appRev.Status.Succeeded = succeed
+	appRev.Status.Succeeded = wfStatus.Phase == workflowv1alpha1.WorkflowStateSucceeded
 	appRev.Status.Workflow = wfStatus
 
 	// Versioned the context backend values.