github/kubevela
1
0
Fork 0
mirror of https://github.com/kubevela/kubevela.git synced 2026-08-19 04:26:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

Feat: Record workflow execution state (#2479)

Browse Source 
* Feat: workflow status add timestamp

* Feat: add workflow recorder

* Fix: finish state process

* Fix: test cases

* Fix: revert comments history->recorder

* Feat: omitempty execution time

* Fix: workflow_test.go

* Feat: add application test case for record

* Feat: terminate record

* Feat: trigger only by anno

* Feat: remove initializers CRD
This commit is contained in:
Jian.Li
2021-10-19 11:43:06 +08:00
committed by GitHub
parent dcb677f976
commit 9fea82fd7b
20 changed files with 640 additions and 1322 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apis/core.oam.dev/common/types.go
+11 -1
View File
@@ -20,6 +20,7 @@ import (
"encoding/json"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
types "github.com/oam-dev/terraform-controller/api/types/crossplane-runtime"
@@ -198,7 +199,9 @@ const (
WorkflowStateTerminated WorkflowState = "terminated"
// WorkflowStateSuspended means workflow is suspended manually, and it can be resumed.
WorkflowStateSuspended WorkflowState = "suspended"
// WorkflowStateFinished means workflow is running successfully, all steps finished.
// WorkflowStateSucceeded means workflow is running successfully, all steps finished.
WorkflowStateSucceeded WorkflowState = "Succeeded"
// WorkflowStateFinished means workflow is end.
WorkflowStateFinished WorkflowState = "finished"
// WorkflowStateExecuting means workflow is still running or waiting some steps.
WorkflowStateExecuting WorkflowState = "executing"
@@ -250,6 +253,10 @@ type WorkflowStepStatus struct {
// A brief CamelCase message indicating details about why the workflowStep is in this state.
Reason string `json:"reason,omitempty"`
SubSteps *SubStepsStatus `json:"subSteps,omitempty"`
// FirstExecuteTime is the first time this step execution.
FirstExecuteTime metav1.Time `json:"firstExecuteTime,omitempty"`
// LastExecuteTime is the last time this step execution.
LastExecuteTime metav1.Time `json:"lastExecuteTime,omitempty"`
}
// WorkflowSubStepStatus record the status of a workflow step
@@ -305,9 +312,12 @@ type WorkflowStatus struct {
Suspend bool `json:"suspend"`
Terminated bool `json:"terminated"`
Finished bool `json:"finished"`
ContextBackend *corev1.ObjectReference `json:"contextBackend,omitempty"`
Steps []WorkflowStepStatus `json:"steps,omitempty"`
StartTime metav1.Time `json:"startTime,omitempty"`
}
// SubStepsStatus record the status of workflow steps.
apis/core.oam.dev/common/zz_generated.deepcopy.go
+3
View File
@@ -595,6 +595,7 @@ func (in *WorkflowStatus) DeepCopyInto(out *WorkflowStatus) {
(*in)[i].DeepCopyInto(&(*out)[i])
}
}
in.StartTime.DeepCopyInto(&out.StartTime)
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new WorkflowStatus.
@@ -615,6 +616,8 @@ func (in *WorkflowStepStatus) DeepCopyInto(out *WorkflowStepStatus) {
*out = new(SubStepsStatus)
(*in).DeepCopyInto(*out)
}
in.FirstExecuteTime.DeepCopyInto(&out.FirstExecuteTime)
in.LastExecuteTime.DeepCopyInto(&out.LastExecuteTime)
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new WorkflowStepStatus.
charts/vela-core/crds/core.oam.dev_applicationrevisions.yaml
+32
View File
@@ -966,16 +966,31 @@ spec:
description: 'UID of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#uids'
type: string
type: object
finished:
type: boolean
mode:
description: WorkflowMode describes the mode of workflow
type: string
startTime:
format: date-time
type: string
steps:
items:
description: WorkflowStepStatus record the status of
a workflow step
properties:
firstExecuteTime:
description: FirstExecuteTime is the first time
this step execution.
format: date-time
type: string
id:
type: string
lastExecuteTime:
description: LastExecuteTime is the last time this
step execution.
format: date-time
type: string
message:
description: A human readable message indicating
details about why the workflowStep is in this
@@ -1043,6 +1058,7 @@ spec:
terminated:
type: boolean
required:
- finished
- mode
- suspend
- terminated
@@ -3094,16 +3110,31 @@ spec:
description: 'UID of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#uids'
type: string
type: object
finished:
type: boolean
mode:
description: WorkflowMode describes the mode of workflow
type: string
startTime:
format: date-time
type: string
steps:
items:
description: WorkflowStepStatus record the status of
a workflow step
properties:
firstExecuteTime:
description: FirstExecuteTime is the first time
this step execution.
format: date-time
type: string
id:
type: string
lastExecuteTime:
description: LastExecuteTime is the last time this
step execution.
format: date-time
type: string
message:
description: A human readable message indicating
details about why the workflowStep is in this
@@ -3171,6 +3202,7 @@ spec:
terminated:
type: boolean
required:
- finished
- mode
- suspend
- terminated
charts/vela-core/crds/core.oam.dev_applications.yaml
+28
View File
@@ -648,15 +648,28 @@ spec:
description: 'UID of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#uids'
type: string
type: object
finished:
type: boolean
mode:
description: WorkflowMode describes the mode of workflow
type: string
startTime:
format: date-time
type: string
steps:
items:
description: WorkflowStepStatus record the status of a workflow step
properties:
firstExecuteTime:
description: FirstExecuteTime is the first time this step execution.
format: date-time
type: string
id:
type: string
lastExecuteTime:
description: LastExecuteTime is the last time this step execution.
format: date-time
type: string
message:
description: A human readable message indicating details about why the workflowStep is in this state.
type: string
@@ -711,6 +724,7 @@ spec:
terminated:
type: boolean
required:
- finished
- mode
- suspend
- terminated
@@ -1441,15 +1455,28 @@ spec:
description: 'UID of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#uids'
type: string
type: object
finished:
type: boolean
mode:
description: WorkflowMode describes the mode of workflow
type: string
startTime:
format: date-time
type: string
steps:
items:
description: WorkflowStepStatus record the status of a workflow step
properties:
firstExecuteTime:
description: FirstExecuteTime is the first time this step execution.
format: date-time
type: string
id:
type: string
lastExecuteTime:
description: LastExecuteTime is the last time this step execution.
format: date-time
type: string
message:
description: A human readable message indicating details about why the workflowStep is in this state.
type: string
@@ -1504,6 +1531,7 @@ spec:
terminated:
type: boolean
required:
- finished
- mode
- suspend
- terminated
charts/vela-core/crds/core.oam.dev_initializers.yaml
-1303
View File
File diff suppressed because it is too large Load Diff
charts/vela-minimal/crds/core.oam.dev_applicationrevisions.yaml
+32
View File
@@ -966,16 +966,31 @@ spec:
description: 'UID of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#uids'
type: string
type: object
finished:
type: boolean
mode:
description: WorkflowMode describes the mode of workflow
type: string
startTime:
format: date-time
type: string
steps:
items:
description: WorkflowStepStatus record the status of
a workflow step
properties:
firstExecuteTime:
description: FirstExecuteTime is the first time
this step execution.
format: date-time
type: string
id:
type: string
lastExecuteTime:
description: LastExecuteTime is the last time this
step execution.
format: date-time
type: string
message:
description: A human readable message indicating
details about why the workflowStep is in this
@@ -1043,6 +1058,7 @@ spec:
terminated:
type: boolean
required:
- finished
- mode
- suspend
- terminated
@@ -3094,16 +3110,31 @@ spec:
description: 'UID of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#uids'
type: string
type: object
finished:
type: boolean
mode:
description: WorkflowMode describes the mode of workflow
type: string
startTime:
format: date-time
type: string
steps:
items:
description: WorkflowStepStatus record the status of
a workflow step
properties:
firstExecuteTime:
description: FirstExecuteTime is the first time
this step execution.
format: date-time
type: string
id:
type: string
lastExecuteTime:
description: LastExecuteTime is the last time this
step execution.
format: date-time
type: string
message:
description: A human readable message indicating
details about why the workflowStep is in this
@@ -3171,6 +3202,7 @@ spec:
terminated:
type: boolean
required:
- finished
- mode
- suspend
- terminated
charts/vela-minimal/crds/core.oam.dev_applications.yaml
+28
View File
@@ -648,15 +648,28 @@ spec:
description: 'UID of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#uids'
type: string
type: object
finished:
type: boolean
mode:
description: WorkflowMode describes the mode of workflow
type: string
startTime:
format: date-time
type: string
steps:
items:
description: WorkflowStepStatus record the status of a workflow step
properties:
firstExecuteTime:
description: FirstExecuteTime is the first time this step execution.
format: date-time
type: string
id:
type: string
lastExecuteTime:
description: LastExecuteTime is the last time this step execution.
format: date-time
type: string
message:
description: A human readable message indicating details about why the workflowStep is in this state.
type: string
@@ -711,6 +724,7 @@ spec:
terminated:
type: boolean
required:
- finished
- mode
- suspend
- terminated
@@ -1441,15 +1455,28 @@ spec:
description: 'UID of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#uids'
type: string
type: object
finished:
type: boolean
mode:
description: WorkflowMode describes the mode of workflow
type: string
startTime:
format: date-time
type: string
steps:
items:
description: WorkflowStepStatus record the status of a workflow step
properties:
firstExecuteTime:
description: FirstExecuteTime is the first time this step execution.
format: date-time
type: string
id:
type: string
lastExecuteTime:
description: LastExecuteTime is the last time this step execution.
format: date-time
type: string
message:
description: A human readable message indicating details about why the workflowStep is in this state.
type: string
@@ -1504,6 +1531,7 @@ spec:
terminated:
type: boolean
required:
- finished
- mode
- suspend
- terminated
legacy/charts/vela-core-legacy/crds/core.oam.dev_applicationrevisions.yaml
+32
View File
@@ -966,16 +966,31 @@ spec:
description: 'UID of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#uids'
type: string
type: object
finished:
type: boolean
mode:
description: WorkflowMode describes the mode of workflow
type: string
startTime:
format: date-time
type: string
steps:
items:
description: WorkflowStepStatus record the status of
a workflow step
properties:
firstExecuteTime:
description: FirstExecuteTime is the first time
this step execution.
format: date-time
type: string
id:
type: string
lastExecuteTime:
description: LastExecuteTime is the last time this
step execution.
format: date-time
type: string
message:
description: A human readable message indicating
details about why the workflowStep is in this
@@ -1043,6 +1058,7 @@ spec:
terminated:
type: boolean
required:
- finished
- mode
- suspend
- terminated
@@ -3094,16 +3110,31 @@ spec:
description: 'UID of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#uids'
type: string
type: object
finished:
type: boolean
mode:
description: WorkflowMode describes the mode of workflow
type: string
startTime:
format: date-time
type: string
steps:
items:
description: WorkflowStepStatus record the status of
a workflow step
properties:
firstExecuteTime:
description: FirstExecuteTime is the first time
this step execution.
format: date-time
type: string
id:
type: string
lastExecuteTime:
description: LastExecuteTime is the last time this
step execution.
format: date-time
type: string
message:
description: A human readable message indicating
details about why the workflowStep is in this
@@ -3171,6 +3202,7 @@ spec:
terminated:
type: boolean
required:
- finished
- mode
- suspend
- terminated
legacy/charts/vela-core-legacy/crds/core.oam.dev_applications.yaml
+32
View File
@@ -882,16 +882,31 @@ spec:
description: 'UID of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#uids'
type: string
type: object
finished:
type: boolean
mode:
description: WorkflowMode describes the mode of workflow
type: string
startTime:
format: date-time
type: string
steps:
items:
description: WorkflowStepStatus record the status of a workflow
step
properties:
firstExecuteTime:
description: FirstExecuteTime is the first time this step
execution.
format: date-time
type: string
id:
type: string
lastExecuteTime:
description: LastExecuteTime is the last time this step
execution.
format: date-time
type: string
message:
description: A human readable message indicating details
about why the workflowStep is in this state.
@@ -956,6 +971,7 @@ spec:
terminated:
type: boolean
required:
- finished
- mode
- suspend
- terminated
@@ -1940,16 +1956,31 @@ spec:
description: 'UID of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#uids'
type: string
type: object
finished:
type: boolean
mode:
description: WorkflowMode describes the mode of workflow
type: string
startTime:
format: date-time
type: string
steps:
items:
description: WorkflowStepStatus record the status of a workflow
step
properties:
firstExecuteTime:
description: FirstExecuteTime is the first time this step
execution.
format: date-time
type: string
id:
type: string
lastExecuteTime:
description: LastExecuteTime is the last time this step
execution.
format: date-time
type: string
message:
description: A human readable message indicating details
about why the workflowStep is in this state.
@@ -2014,6 +2045,7 @@ spec:
terminated:
type: boolean
required:
- finished
- mode
- suspend
- terminated
legacy/charts/vela-core-legacy/crds/core.oam.dev_initializers.yaml
+16
View File
@@ -1085,16 +1085,31 @@ spec:
description: 'UID of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#uids'
type: string
type: object
finished:
type: boolean
mode:
description: WorkflowMode describes the mode of workflow
type: string
startTime:
format: date-time
type: string
steps:
items:
description: WorkflowStepStatus record the status of
a workflow step
properties:
firstExecuteTime:
description: FirstExecuteTime is the first time
this step execution.
format: date-time
type: string
id:
type: string
lastExecuteTime:
description: LastExecuteTime is the last time this
step execution.
format: date-time
type: string
message:
description: A human readable message indicating
details about why the workflowStep is in this
@@ -1162,6 +1177,7 @@ spec:
terminated:
type: boolean
required:
- finished
- mode
- suspend
- terminated
pkg/controller/core.oam.dev/v1alpha2/application/application_controller.go
+26 -5
View File
@@ -175,7 +175,8 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
return r.endWithNegativeCondition(ctx, app, condition.ErrorCondition("Workflow", err), common.ApplicationRunningWorkflow)
}
workflowState, err := workflow.NewWorkflow(app, r.Client, appFile.WorkflowMode).ExecuteSteps(ctx, handler.currentAppRev, steps)
wf := workflow.NewWorkflow(app, r.Client, appFile.WorkflowMode)
workflowState, err := wf.ExecuteSteps(ctx, handler.currentAppRev, steps)
if err != nil {
klog.Error(err, "[handle workflow]")
r.Recorder.Event(app, event.Warning(velatypes.ReasonFailedWorkflow, err))
@@ -185,14 +186,18 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
handler.addServiceStatus(false, app.Status.Services...)
handler.addAppliedResource(app.Status.AppliedResources...)
app.Status.AppliedResources = handler.appliedResources
app.Status.Services = handler.services
switch workflowState {
case common.WorkflowStateSuspended:
return ctrl.Result{}, r.patchStatus(ctx, app, common.ApplicationWorkflowSuspending)
case common.WorkflowStateTerminated:
if err := r.doWorkflowFinish(app, wf); err != nil {
return r.endWithNegativeCondition(ctx, app, condition.ErrorCondition("DoWorkflowFinish", err), common.ApplicationRunningWorkflow)
}
return ctrl.Result{}, r.patchStatus(ctx, app, common.ApplicationWorkflowTerminated)
case common.WorkflowStateExecuting:
return reconcile.Result{RequeueAfter: baseWorkflowBackoffWaitTime}, r.patchStatus(ctx, app, common.ApplicationRunningWorkflow)
case common.WorkflowStateFinished:
case common.WorkflowStateSucceeded:
wfStatus := app.Status.Workflow
if wfStatus != nil {
ref, err := handler.DispatchAndGC(ctx)
@@ -210,10 +215,18 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
}
app.Status.ResourceTracker = ref
}
if err := r.doWorkflowFinish(app, wf); err != nil {
return r.endWithNegativeCondition(ctx, app, condition.ErrorCondition("DoWorkflowFinish", err), common.ApplicationRunningWorkflow)
}
app.Status.SetConditions(condition.ReadyCondition("WorkflowFinished"))
r.Recorder.Event(app, event.Normal(velatypes.ReasonApplied, velatypes.MessageWorkflowFinished))
klog.Info("Application manifests has applied by workflow successfully", "application", klog.KObj(app))
return ctrl.Result{}, r.patchStatus(ctx, app, common.ApplicationWorkflowFinished)
case common.WorkflowStateFinished:
if status := app.Status.Workflow; status != nil && status.Terminated {
return ctrl.Result{}, nil
}
}
app.Status.SetConditions(condition.ReadyCondition("WorkflowFinished"))
r.Recorder.Event(app, event.Normal(velatypes.ReasonApplied, velatypes.MessageWorkflowFinished))
klog.Info("Application manifests has applied by workflow successfully", "application", klog.KObj(app))
} else {
var comps []*velatypes.ComponentManifest
comps, err = appFile.GenerateComponentManifests()
@@ -342,6 +355,14 @@ func (r *Reconciler) patchStatus(ctx context.Context, app *v1beta1.Application,
return r.Client.Status().Patch(ctx, app, client.Merge)
}
func (r *Reconciler) doWorkflowFinish(app *v1beta1.Application, wf workflow.Workflow) error {
if err := wf.Trace(); err != nil {
return errors.WithMessage(err, "record workflow state")
}
app.Status.Workflow.Finished = true
return nil
}
// appWillRollout judge whether the application will be released by rollout.
// If it's true, application controller will only create or update application revision but not emit any other K8s
// resources into the cluster. Rollout controller will do real release works.
pkg/controller/core.oam.dev/v1alpha2/application/application_controller_test.go
+52 -2
View File
@@ -26,8 +26,6 @@ import (
"strconv"
"time"
"github.com/oam-dev/kubevela/pkg/oam/testutil"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
@@ -52,8 +50,10 @@ import (
velatypes "github.com/oam-dev/kubevela/apis/types"
"github.com/oam-dev/kubevela/pkg/controller/utils"
"github.com/oam-dev/kubevela/pkg/oam"
"github.com/oam-dev/kubevela/pkg/oam/testutil"
"github.com/oam-dev/kubevela/pkg/oam/util"
common2 "github.com/oam-dev/kubevela/pkg/utils/common"
wfTypes "github.com/oam-dev/kubevela/pkg/workflow/types"
)
// TODO: Refactor the tests to not copy and paste duplicated code 10 times
@@ -1465,6 +1465,7 @@ var _ = Describe("Test Application Controller", func() {
checkApp.Spec.Components[0].Properties = &runtime.RawExtension{Raw: []byte(`{"cmd":["sleep","2000"],"image":"nginx"}`)}
Expect(k8sClient.Update(ctx, checkApp)).Should(BeNil())
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.ApplicationRunning))
@@ -1480,6 +1481,7 @@ var _ = Describe("Test Application Controller", func() {
checkApp.Spec.Components[0].Traits[0].Properties = &runtime.RawExtension{Raw: []byte(`{"targetRevision":"myweb1-v3"}`)}
Expect(k8sClient.Update(ctx, checkApp)).Should(BeNil())
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.LatestRevision.Name).Should(BeEquivalentTo(appRevName))
@@ -1533,6 +1535,7 @@ var _ = Describe("Test Application Controller", func() {
testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey})
// second reconcile apply all resources
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.ApplicationRunning))
@@ -1579,6 +1582,7 @@ var _ = Describe("Test Application Controller", func() {
testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey})
// second reconcile apply all resources
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.ApplicationRunning))
@@ -1628,6 +1632,7 @@ var _ = Describe("Test Application Controller", func() {
testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey})
// second reconcile apply all resources
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.ApplicationRunning))
@@ -1798,6 +1803,7 @@ var _ = Describe("Test Application Controller", func() {
Expect(expDeployment.Spec.Template.Spec.Containers[0].Command).Should(BeEquivalentTo([]string{"sleep", "100"}))
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.ApplicationRunning))
@@ -1880,6 +1886,7 @@ var _ = Describe("Test Application Controller", func() {
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})
testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey})
checkApp = &v1beta1.Application{}
@@ -1968,6 +1975,7 @@ var _ = Describe("Test Application Controller", func() {
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})
testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey})
checkApp = &v1beta1.Application{}
@@ -2069,6 +2077,48 @@ var _ = Describe("Test Application Controller", func() {
}))
})
It("app record execution state with controllerRevision", func() {
ns := &corev1.Namespace{
ObjectMeta: metav1.ObjectMeta{
Name: "vela-test-app-trace",
},
}
app := appwithNoTrait.DeepCopy()
app.Name = "vela-test-app-trace"
app.SetNamespace(ns.Name)
app.Annotations = map[string]string{wfTypes.AnnotationPublishVersion: "v134"}
Expect(k8sClient.Create(ctx, ns)).Should(BeNil())
Expect(k8sClient.Create(ctx, app)).Should(BeNil())
appKey := client.ObjectKey{
Name: app.Name,
Namespace: app.Namespace,
}
testutil.ReconcileOnceAfterFinalizer(reconciler, reconcile.Request{NamespacedName: appKey})
checkApp := &v1beta1.Application{}
Expect(k8sClient.Get(ctx, appKey, checkApp)).Should(BeNil())
Expect(checkApp.Status.Phase).Should(BeEquivalentTo(common.ApplicationRunning))
recorder := &v1.ControllerRevision{}
Expect(k8sClient.Get(ctx, client.ObjectKey{
Name: fmt.Sprintf("record-%s-v134", app.Name),
Namespace: app.Namespace,
}, recorder)).Should(BeNil())
checkApp.Annotations[wfTypes.AnnotationPublishVersion] = "v135"
Expect(k8sClient.Update(ctx, checkApp)).Should(BeNil())
testutil.ReconcileOnceAfterFinalizer(reconciler, reconcile.Request{NamespacedName: appKey})
checkApp = &v1beta1.Application{}
Expect(k8sClient.Get(ctx, appKey, checkApp)).Should(BeNil())
Expect(checkApp.Status.Phase).Should(BeEquivalentTo(common.ApplicationRunning))
Expect(k8sClient.Get(ctx, client.ObjectKey{
Name: fmt.Sprintf("record-%s-v135", app.Name),
Namespace: app.Namespace,
}, recorder)).Should(BeNil())
})
})
const (
pkg/controller/core.oam.dev/v1alpha2/application/workflow_test.go
+4
View File
@@ -218,6 +218,7 @@ var _ = Describe("Test Workflow", func() {
triggerWorkflowStepToSucceed(stepObj)
Expect(k8sClient.Update(ctx, stepObj)).Should(BeNil())
tryReconcile(reconciler, appWithWorkflow.Name, appWithWorkflow.Namespace)
tryReconcile(reconciler, appWithWorkflow.Name, appWithWorkflow.Namespace)
tryReconcile(reconciler, appWithWorkflow.Name, appWithWorkflow.Namespace)
@@ -416,6 +417,7 @@ var _ = Describe("Test Workflow", func() {
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})
testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey})
checkApp = &oamcore.Application{}
@@ -513,6 +515,7 @@ var _ = Describe("Test Workflow", func() {
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})
testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey})
checkApp = &oamcore.Application{}
@@ -614,6 +617,7 @@ var _ = Describe("Test Workflow", func() {
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})
testutil.ReconcileOnce(reconciler, reconcile.Request{NamespacedName: appKey})
checkApp = &oamcore.Application{}
pkg/oam/testutil/helper.go
+3
View File
@@ -61,6 +61,9 @@ func ReconcileOnceAfterFinalizer(r reconcile.Reconciler, req reconcile.Request)
if result, err := r.Reconcile(context.TODO(), req); err != nil {
return result, err
}
if result, err := r.Reconcile(context.TODO(), req); err != nil {
return result, err
}
return r.Reconcile(context.TODO(), req)
}
pkg/workflow/interface.go
+2
View File
@@ -28,4 +28,6 @@ type Workflow interface {
// ExecuteSteps executes the steps of an Application with given steps of rendered resources.
// It returns done=true only if all steps are executed and succeeded.
ExecuteSteps(ctx context.Context, appRev *v1beta1.ApplicationRevision, taskRunners []types.TaskRunner) (state common.WorkflowState, err error)
// Trace record workflow state in controllerRevision.
Trace() error
}
pkg/workflow/recorder/recorder.go
+156
View File
@@ -0,0 +1,156 @@
/*Copyright 2021 The KubeVela Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package recorder
import (
"context"
"fmt"
"sort"
"strings"
"time"
"github.com/pkg/errors"
apps "k8s.io/api/apps/v1"
kerrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/runtime"
"sigs.k8s.io/controller-runtime/pkg/client"
"github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1"
)
const (
// LabelRecordSource is label that describe recorder source.
LabelRecordSource = "vela.io/source"
// LabelRecordVersion is label that describe recorder version.
LabelRecordVersion = "vela.io/wf-revision"
)
type recorder struct {
cli client.Client
source *v1beta1.Application
err error
}
// With a recorder store.
func With(cli client.Client, source *v1beta1.Application) Store {
return &recorder{
cli: cli,
source: source,
}
}
// Save object to controllerRevision.
func (r *recorder) Save(version string, data []byte) Store {
if r.err != nil {
return r
}
rv := &apps.ControllerRevision{}
rv.Namespace = r.source.GetNamespace()
rv.Revision = time.Now().UnixNano()
if version == "" {
wfStatus := r.source.Status.Workflow
if wfStatus != nil {
if !strings.Contains(wfStatus.AppRevision, ":") {
version = wfStatus.AppRevision
}
}
}
if version == "" {
version = fmt.Sprint(rv.Revision)
}
rv.Name = fmt.Sprintf("record-%s-%s", r.source.Name, version)
rv.SetLabels(map[string]string{
LabelRecordSource: r.source.GetName(),
LabelRecordVersion: version,
})
ownerRef := metav1.NewControllerRef(r.source, r.source.GroupVersionKind())
ownerRef.APIVersion = v1beta1.SchemeGroupVersion.String()
ownerRef.Kind = v1beta1.ApplicationKind
rv.SetOwnerReferences([]metav1.OwnerReference{*ownerRef})
rv.Data = runtime.RawExtension{
Raw: data,
}
if err := r.cli.Create(context.Background(), rv); err != nil && !kerrors.IsAlreadyExists(err) {
r.err = errors.WithMessagef(err, "save record %s/%s", rv.Namespace, rv.Name)
}
return r
}
// Limit gc over limit controllerRevisions.
func (r *recorder) Limit(max int) Store {
if r.err != nil {
return r
}
selector, err := labels.Parse(fmt.Sprintf("%s=%s", LabelRecordSource, r.source.GetName()))
if err != nil {
r.err = errors.WithMessagef(err, "limit recorder: make selector(source=%s)", r.source.GetName())
return r
}
rds := &apps.ControllerRevisionList{}
if err := r.cli.List(context.Background(), rds, &client.ListOptions{
LabelSelector: selector,
}); err != nil {
r.err = errors.WithMessagef(err, "limit recorder: list controllerRevision (source=%s)", r.source.GetName())
}
if len(rds.Items) <= max {
return r
}
items := rds.Items
sort.Sort(rvSorter(items))
for i := 0; i < len(rds.Items)-max; i++ {
o := &items[i]
if err := r.cli.Delete(context.Background(), o); err != nil {
r.err = errors.WithMessage(err, "limit recorder: delete controllerRevision")
break
}
}
return r
}
// Error return error info.
func (r *recorder) Error() error {
return r.err
}
// Store is an object that record info.
type Store interface {
Save(version string, data []byte) Store
Limit(max int) Store
Error() error
}
type rvSorter []apps.ControllerRevision
// Len is the number of elements in the collection.
func (rs rvSorter) Len() int {
return len(rs)
}
// Less reports whether the element with index i must sort before the element with index j.
func (rs rvSorter) Less(i, j int) bool {
return rs[i].Revision < rs[j].Revision
}
// Swap swaps the elements with indexes i and j.
func (rs rvSorter) Swap(i, j int) {
rs[i], rs[j] = rs[j], rs[i]
}
pkg/workflow/recorder/recorder_test.go
+113
View File
@@ -0,0 +1,113 @@
/*Copyright 2021 The KubeVela Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package recorder
import (
"context"
"fmt"
"testing"
"github.com/crossplane/crossplane-runtime/pkg/test"
"github.com/pkg/errors"
"gotest.tools/assert"
apps "k8s.io/api/apps/v1"
"sigs.k8s.io/controller-runtime/pkg/client"
"github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1"
)
func TestRecord(t *testing.T) {
cli := makeMockClient()
app := &v1beta1.Application{}
app.Namespace = "default"
app.Name = "test-app"
data := []byte(`{"spec": "mock-data"}`)
err := With(cli, app).Save("v1", data).
Save("v2", data).
Save("v3", data).Limit(2).Error()
assert.NilError(t, err)
crs := &apps.ControllerRevisionList{}
err = cli.List(context.Background(), crs)
assert.NilError(t, err)
assert.Equal(t, len(crs.Items), 2)
assert.Equal(t, crs.Items[0].Name, "record-test-app-v2")
assert.Equal(t, crs.Items[1].Name, "record-test-app-v3")
creatErrorEnable = true
err = With(cli, app).Save("v1", data).Error()
assert.Equal(t, err.Error(), "save record default/record-test-app-v1: mock create error")
creatErrorEnable = false
listErrorEnable = true
err = With(cli, app).Save("v1", data).Limit(3).Error()
assert.Equal(t, err.Error(), "limit recorder: list controllerRevision (source=test-app): mock list error")
listErrorEnable = false
cli = makeMockClient()
err = With(cli, app).Save("", data).Limit(1).Error()
assert.NilError(t, err)
crs = &apps.ControllerRevisionList{}
err = cli.List(context.Background(), crs)
assert.NilError(t, err)
assert.Equal(t, crs.Items[0].Name, fmt.Sprintf("record-%s-%d", app.Name, crs.Items[0].Revision))
}
var (
listErrorEnable bool
creatErrorEnable bool
)
func makeMockClient() client.Client {
items := []apps.ControllerRevision{}
return &test.MockClient{
MockList: func(ctx context.Context, list client.ObjectList, opts ...client.ListOption) error {
if listErrorEnable {
return errors.New("mock list error")
}
crList, ok := list.(*apps.ControllerRevisionList)
if ok {
*crList = apps.ControllerRevisionList{
Items: items,
}
}
return nil
},
MockCreate: func(ctx context.Context, obj client.Object, opts ...client.CreateOption) error {
if creatErrorEnable {
return errors.New("mock create error")
}
o, ok := obj.(*apps.ControllerRevision)
if ok {
items = append(items, *o)
}
return nil
},
MockDelete: func(ctx context.Context, obj client.Object, opts ...client.DeleteOption) error {
o, ok := obj.(*apps.ControllerRevision)
if ok {
newItems := []apps.ControllerRevision{}
for index := range items {
if items[index].Name != o.Name || items[index].Namespace != o.Namespace {
newItems = append(newItems, items[index])
}
}
items = newItems
}
return nil
},
}
}
pkg/workflow/types/types.go
+2
View File
@@ -77,4 +77,6 @@ type Action interface {
const (
// ContextKeyMetadata is key that refer to application metadata.
ContextKeyMetadata = "metadata__"
// AnnotationPublishVersion is annotation that record the application workflow version.
AnnotationPublishVersion = "vela.io/publish-version"
)
pkg/workflow/workflow.go
+44 -7
View File
@@ -18,9 +18,12 @@ package workflow
import (
"context"
"encoding/json"
"fmt"
"time"
"github.com/pkg/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"sigs.k8s.io/controller-runtime/pkg/client"
"github.com/oam-dev/kubevela/apis/core.oam.dev/common"
@@ -29,6 +32,7 @@ import (
"github.com/oam-dev/kubevela/pkg/cue/model/value"
"github.com/oam-dev/kubevela/pkg/oam/util"
wfContext "github.com/oam-dev/kubevela/pkg/workflow/context"
"github.com/oam-dev/kubevela/pkg/workflow/recorder"
wfTypes "github.com/oam-dev/kubevela/pkg/workflow/types"
)
@@ -62,29 +66,37 @@ func (w *workflow) ExecuteSteps(ctx context.Context, appRev *oamcore.Application
}
if w.app.Status.Workflow == nil || w.app.Status.Workflow.AppRevision != revAndSpecHash {
status := w.app.Status.Workflow
if status != nil && !status.Finished {
status.Terminated = true
return common.WorkflowStateTerminated, nil
}
w.app.Status.Workflow = &common.WorkflowStatus{
AppRevision: revAndSpecHash,
Mode: common.WorkflowModeStep,
StartTime: metav1.NewTime(time.Now()),
}
if w.dagMode {
w.app.Status.Workflow.Mode = common.WorkflowModeDAG
}
// clean recorded resources info.
w.app.Status.Services = nil
w.app.Status.AppliedResources = nil
}
wfStatus := w.app.Status.Workflow
allTasksDone := w.allDone(taskRunners)
if wfStatus.Finished {
return common.WorkflowStateFinished, nil
}
if wfStatus.Terminated {
return common.WorkflowStateTerminated, nil
}
if wfStatus.Suspend {
return common.WorkflowStateSuspended, nil
}
allTasksDone := w.allDone(taskRunners)
if allTasksDone {
return common.WorkflowStateFinished, nil
return common.WorkflowStateSucceeded, nil
}
var (
@@ -112,11 +124,20 @@ func (w *workflow) ExecuteSteps(ctx context.Context, appRev *oamcore.Application
return common.WorkflowStateSuspended, nil
}
if w.allDone(taskRunners) {
return common.WorkflowStateFinished, nil
return common.WorkflowStateSucceeded, nil
}
return common.WorkflowStateExecuting, nil
}
// Trace record the workflow execute history.
func (w *workflow) Trace() error {
data, err := json.Marshal(w.app)
if err != nil {
return err
}
return recorder.With(w.cli, w.app).Save("", data).Limit(10).Error()
}
func (w *workflow) allDone(taskRunners []wfTypes.TaskRunner) bool {
status := w.app.Status.Workflow
for _, t := range taskRunners {
@@ -285,15 +306,21 @@ func (e *engine) finishStep(operation *wfTypes.Operation) {
}
func (e *engine) updateStepStatus(status common.WorkflowStepStatus) {
var conditionUpdated bool
var (
conditionUpdated bool
now = metav1.NewTime(time.Now())
)
status.LastExecuteTime = now
for i := range e.status.Steps {
if e.status.Steps[i].Name == status.Name {
status.FirstExecuteTime = e.status.Steps[i].FirstExecuteTime
e.status.Steps[i] = status
conditionUpdated = true
break
}
}
if !conditionUpdated {
status.FirstExecuteTime = now
e.status.Steps = append(e.status.Steps, status)
}
}
@@ -303,6 +330,16 @@ func (e *engine) needStop() bool {
}
func computeAppRevisionHash(rev string, app *oamcore.Application) (string, error) {
specHash, err := utils.ComputeSpecHash(app.Spec)
return fmt.Sprintf("%s:%s", rev, specHash), err
version := ""
if annos := app.Annotations; annos != nil {
version = annos[wfTypes.AnnotationPublishVersion]
}
if version == "" {
specHash, err := utils.ComputeSpecHash(app.Spec)
if err != nil {
return "", err
}
version = fmt.Sprintf("%s:%s", rev, specHash)
}
return version, nil
}
pkg/workflow/workflow_test.go
+24 -4
View File
@@ -72,6 +72,7 @@ var _ = Describe("Test Workflow", func() {
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,
@@ -105,8 +106,13 @@ var _ = Describe("Test Workflow", func() {
wf = NewWorkflow(app, k8sClient, common.WorkflowModeStep)
state, err = wf.ExecuteSteps(context.Background(), revision, runners)
Expect(err).ToNot(HaveOccurred())
Expect(state).Should(BeEquivalentTo(common.WorkflowStateFinished))
Expect(state).Should(BeEquivalentTo(common.WorkflowStateTerminated))
app.Status.Workflow.Finished = true
state, err = wf.ExecuteSteps(context.Background(), revision, runners)
Expect(err).ToNot(HaveOccurred())
Expect(state).Should(BeEquivalentTo(common.WorkflowStateSucceeded))
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,
@@ -148,6 +154,7 @@ var _ = Describe("Test Workflow", func() {
Expect(state).Should(BeEquivalentTo(common.WorkflowStateSuspended))
wfStatus := *app.Status.Workflow
wfStatus.ContextBackend = nil
cleanStepTimeStamp(&wfStatus)
Expect(cmp.Diff(wfStatus, common.WorkflowStatus{
AppRevision: wfStatus.AppRevision,
Mode: common.WorkflowModeStep,
@@ -174,8 +181,9 @@ var _ = Describe("Test Workflow", func() {
app.Labels = map[string]string{"for-test": "changed"}
state, err = wf.ExecuteSteps(context.Background(), revision, runners)
Expect(err).ToNot(HaveOccurred())
Expect(state).Should(BeEquivalentTo(common.WorkflowStateFinished))
Expect(state).Should(BeEquivalentTo(common.WorkflowStateSucceeded))
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,
@@ -196,7 +204,7 @@ var _ = Describe("Test Workflow", func() {
state, err = wf.ExecuteSteps(context.Background(), revision, runners)
Expect(err).ToNot(HaveOccurred())
Expect(state).Should(BeEquivalentTo(common.WorkflowStateFinished))
Expect(state).Should(BeEquivalentTo(common.WorkflowStateSucceeded))
})
It("test for terminate", func() {
@@ -215,6 +223,7 @@ var _ = Describe("Test Workflow", func() {
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,
@@ -251,6 +260,7 @@ var _ = Describe("Test Workflow", func() {
Expect(err).To(HaveOccurred())
Expect(state).Should(BeEquivalentTo(common.WorkflowStateExecuting))
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,
@@ -291,6 +301,7 @@ var _ = Describe("Test Workflow", func() {
Expect(err).ToNot(HaveOccurred())
Expect(state).Should(BeEquivalentTo(common.WorkflowStateExecuting))
app.Status.Workflow.ContextBackend = nil
cleanStepTimeStamp(app.Status.Workflow)
Expect(cmp.Diff(*app.Status.Workflow, common.WorkflowStatus{
AppRevision: app.Status.Workflow.AppRevision,
Mode: common.WorkflowModeDAG,
@@ -312,8 +323,9 @@ var _ = Describe("Test Workflow", func() {
pending = false
state, err = wf.ExecuteSteps(context.Background(), revision, runners)
Expect(err).ToNot(HaveOccurred())
Expect(state).Should(BeEquivalentTo(common.WorkflowStateFinished))
Expect(state).Should(BeEquivalentTo(common.WorkflowStateSucceeded))
app.Status.Workflow.ContextBackend = nil
cleanStepTimeStamp(app.Status.Workflow)
Expect(cmp.Diff(*app.Status.Workflow, common.WorkflowStatus{
AppRevision: app.Status.Workflow.AppRevision,
Mode: common.WorkflowModeDAG,
@@ -463,3 +475,11 @@ func (tr *testTaskRunner) Run(ctx wfContext.Context, options *wfTypes.TaskRunOpt
func (tr *testTaskRunner) Pending(ctx wfContext.Context) bool {
return tr.checkPending(ctx)
}
func cleanStepTimeStamp(wfStatus *common.WorkflowStatus) {
wfStatus.StartTime = metav1.Time{}
for index := range wfStatus.Steps {
wfStatus.Steps[index].FirstExecuteTime = metav1.Time{}
wfStatus.Steps[index].LastExecuteTime = metav1.Time{}
}
}