From 592f8b8e8fddbc7af45cb6f95cc6052ca69d5f44 Mon Sep 17 00:00:00 2001 From: Tianxin Dong Date: Tue, 14 Mar 2023 22:14:41 +0800 Subject: [PATCH] Fix: stores workflow status in revison if it is restarted (#5604) (#5673) Signed-off-by: FogDong --- .../application/application_controller.go | 13 ++-- .../application_controller_test.go | 70 +++++++++++++++++++ .../v1alpha2/application/generator.go | 67 ++++++++++-------- .../v1alpha2/application/generator_test.go | 26 ++++--- .../v1alpha2/application/revision.go | 4 +- 5 files changed, 132 insertions(+), 48 deletions(-) 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 60a55950d..930f2c21f 100644 --- a/pkg/controller/core.oam.dev/v1alpha2/application/application_controller.go +++ b/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 { 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 b5c1b19fb..d650114ac 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 @@ -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{ diff --git a/pkg/controller/core.oam.dev/v1alpha2/application/generator.go b/pkg/controller/core.oam.dev/v1alpha2/application/generator.go index 876c3dad8..1e1f35dbd 100644 --- a/pkg/controller/core.oam.dev/v1alpha2/application/generator.go +++ b/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 // : will be recognized and reduced into -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, diff --git a/pkg/controller/core.oam.dev/v1alpha2/application/generator_test.go b/pkg/controller/core.oam.dev/v1alpha2/application/generator_test.go index 3d2642797..74b66bc46 100644 --- a/pkg/controller/core.oam.dev/v1alpha2/application/generator_test.go +++ b/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()) }) }) diff --git a/pkg/controller/core.oam.dev/v1alpha2/application/revision.go b/pkg/controller/core.oam.dev/v1alpha2/application/revision.go index 4b996db22..5f008b951 100644 --- a/pkg/controller/core.oam.dev/v1alpha2/application/revision.go +++ b/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.