app controller: reconcile workflow and create res configmap (#1788)

docs/examples/workflow/README.md

@@ -0,0 +1,130 @@
+1. Apply CRD and Definitions:
+
+    ```
+    kubectl apply -f definition.yaml
+    ```
+
+    Check Policy and Workflow definitions:
+
+    ```
+    kubectl get policy
+    kubectl get workflowstep
+    ```
+
+    Output:
+    ```
+    NAME        AGE
+    foopolicy   41s
+
+    NAME    AGE
+    foowf   49s
+    ```
+
+    Check DefinitionRevision:
+
+    ```
+    kubectl get definitionrevision
+    ```
+
+    Output:
+
+    ```
+    NAMESPACE     NAME             REVISION   HASH               TYPE
+    default       foopolicy-v1     1          8c340e1beaf9a3fa   Policy
+    default       foowf-v1         1          83cf4e8246a89afa   WorkflowStep
+    ```
+
+1. Apply Application:
+
+    ```
+    kubectl apply -f app.yaml
+    ```
+
+1. Check workflow status in Application:
+
+    ```
+    kubectl get application first-vela-app -o=jsonpath='{.status.workflow[?(@.name=="my-wf")]}.phase'
+    ```
+
+    Output:
+    ```
+    running
+    ```
+
+1. Check Workflow objects:
+
+    ```
+    kubectl get foo my-wf -o=jsonpath='{.spec.key}'
+    ```
+
+    Output:
+
+    ```
+    test
+    ```
+
+    This means the resource has been rendered correctly.
+
+1. Check workflow context:
+
+    ```
+    kubectl get foo my-wf -o=jsonpath='{.metadata.annotations.app\.oam\.dev/workflow-context}' | jq
+    ```
+
+    Output:
+
+    ```json
+    {
+      "appName": "first-vela-app",
+      "appRevision": "first-vela-app-v1",
+      "workflowIndex": 0,
+      "resourceConfigMap": {
+        "name": "first-vela-app-v1"
+      }
+    }
+    ```
+
+1. Patch condition status on workflow object:
+
+    ```
+    kubectl patch foo my-wf --type merge --patch "$(cat wf-patch.yaml)"
+    ```
+
+    Check workflow object status:
+
+    ```
+    kubectl get foo my-wf -o=jsonpath='{.status.conditions[?(@.type=="workflow-finish")]}' | jq
+    ```
+
+    Output:
+
+    ```json
+    {
+      "message": "{\"observedGeneration\":2}",
+      "reason": "Succeeded",
+      "status": "True",
+      "type": "workflow-finish"
+    }
+    ```
+
+    > Note: The observedGeneration is 2 because the json patch will trigger generation increment.
+
+1.  Check workflow status in Application:
+
+    ```
+    kubectl get application first-vela-app -o=jsonpath='{.status.workflow[?(@.name=="my-wf")]}.phase'
+    ```
+
+    Output:
+    ```
+    succeeded
+    ```
+
+    The workflow phase has changed from running to succeeded due to the underlying object changing status condition.
+
+1. cleanup:
+
+    ```
+    kubectl delete -f app.yaml
+    kubectl delete -f definition.yaml
+    ```

docs/examples/workflow/app.yaml

@@ -0,0 +1,28 @@
+apiVersion: core.oam.dev/v1beta1
+kind: Application
+metadata:
+  name: first-vela-app
+spec:
+  components:
+    - name: express-server
+      type: webservice
+      properties:
+        image: crccheck/hello-world
+        port: 8000
+      traits:
+        - type: ingress
+          properties:
+            domain: testsvc.example.com
+            http:
+              "/": 8000
+  policies:
+    - name: my-policy
+      type: foopolicy
+      properties:
+        key: test
+
+  workflow:
+    - name: my-wf
+      type: foowf
+      properties:
+        key: test

docs/examples/workflow/definition.yaml

@@ -0,0 +1,52 @@
+apiVersion: apiextensions.k8s.io/v1beta1
+kind: CustomResourceDefinition
+metadata:
+  name: foo.example.com
+spec:
+  group: example.com
+  names:
+    kind: Foo
+    listKind: FooList
+    plural: foo
+    singular: foo
+  scope: Namespaced
+  version: v1
+  preserveUnknownFields: true
+---
+apiVersion: core.oam.dev/v1beta1
+kind: PolicyDefinition
+metadata:
+  name: foopolicy
+spec:
+  schematic:
+    cue:
+      template: |
+        output: {
+          apiVersion: "example.com/v1"
+          kind:       "Foo"
+          spec: {
+            key: parameter.key
+          }
+        }
+        parameter: {
+          key: string
+        }
+---
+apiVersion: core.oam.dev/v1beta1
+kind: WorkflowStepDefinition
+metadata:
+  name: foowf
+spec:
+  schematic:
+    cue:
+      template: |
+        output: {
+          apiVersion: "example.com/v1"
+          kind:       "Foo"
+          spec: {
+            key: parameter.key
+          }
+        }
+        parameter: {
+          key: string
+        }

docs/examples/workflow/wf-patch.yaml

@@ -0,0 +1,6 @@
+status:
+  conditions:
+  - type: 'workflow-finish'
+    status: 'True'
+    reason: 'Succeeded'
+    message: '{"observedGeneration":3}'

pkg/appfile/appfile.go

@@ -163,6 +163,42 @@ type Appfile struct {
 	Namespace    string
 	RevisionName string
 	Workloads    []*Workload
+
+	Policies      []*Workload
+	WorkflowSteps []*Workload
+}
+
+// GenerateWorkflowAndPolicy generates workflow steps and policies from an appFile
+func (af *Appfile) GenerateWorkflowAndPolicy() (policies, steps []*unstructured.Unstructured, err error) {
+	policies, err = af.generateUnstructureds(af.Policies)
+	if err != nil {
+		return
+	}
+	steps, err = af.generateUnstructureds(af.WorkflowSteps)
+	if err != nil {
+		return
+	}
+	return
+}
+
+func (af *Appfile) generateUnstructureds(workloads []*Workload) ([]*unstructured.Unstructured, error) {
+	uns := []*unstructured.Unstructured{}
+	for _, wl := range workloads {
+		un, err := generateUnstructuredFromCUEModule(wl, af.Name, af.RevisionName, af.Namespace)
+		if err != nil {
+			return nil, err
+		}
+		uns = append(uns, un)
+	}
+	return uns, nil
+}
+
+func generateUnstructuredFromCUEModule(wl *Workload, appName, revision, ns string) (*unstructured.Unstructured, error) {
+	pCtx, err := PrepareProcessContext(wl, appName, revision, ns)
+	if err != nil {
+		return nil, err
+	}
+	return makeWorkloadWithContext(pCtx, wl, ns, appName)
 }
 
 // GenerateApplicationConfiguration converts an appFile to applicationConfig & Components

pkg/appfile/parser.go

@@ -100,9 +100,45 @@ func (p *Parser) GenerateAppFile(ctx context.Context, app *v1beta1.Application)
 		wds = append(wds, wd)
 	}
 	appfile.Workloads = wds
+
+	var err error
+
+	appfile.Policies, err = p.parsePolicies(ctx, appName, ns, app.Spec.Policies)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parsePolicies: %w", err)
+	}
+
+	appfile.WorkflowSteps, err = p.parseWorkflow(ctx, appName, ns, app.Spec.Workflow)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parseWorkflow: %w", err)
+	}
 	return appfile, nil
 }
 
+func (p *Parser) parsePolicies(ctx context.Context, appName, ns string, policies []v1beta1.AppPolicy) ([]*Workload, error) {
+	ws := []*Workload{}
+	for _, policy := range policies {
+		w, err := p.makeWorkload(ctx, appName, ns, policy.Name, policy.Type, types.TypePolicy, policy.Properties)
+		if err != nil {
+			return nil, err
+		}
+		ws = append(ws, w)
+	}
+	return ws, nil
+}
+
+func (p *Parser) parseWorkflow(ctx context.Context, appName, ns string, steps []v1beta1.WorkflowStep) ([]*Workload, error) {
+	ws := []*Workload{}
+	for _, step := range steps {
+		w, err := p.makeWorkload(ctx, appName, ns, step.Name, step.Type, types.TypeWorkflowStep, step.Properties)
+		if err != nil {
+			return nil, err
+		}
+		ws = append(ws, w)
+	}
+	return ws, nil
+}
+
 func (p *Parser) makeWorkload(ctx context.Context, appName, ns, name, typ string, capType types.CapType, props runtime.RawExtension) (*Workload, error) {
 	templ, err := p.tmplLoader.LoadTemplate(ctx, p.dm, p.client, typ, capType)
 	if err != nil && !kerrors.IsNotFound(err) {
@@ -181,6 +217,7 @@ func (p *Parser) parseWorkload(ctx context.Context, comp v1beta1.ApplicationComp
 	}
 	return workload, nil
 }
+
 func (p *Parser) parseTrait(ctx context.Context, name string, properties map[string]interface{}) (*Trait, error) {
 	templ, err := p.tmplLoader.LoadTemplate(ctx, p.dm, p.client, name, types.TypeTrait)
 	if kerrors.IsNotFound(err) {

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

@@ -26,6 +26,7 @@ import (
 	"github.com/crossplane/crossplane-runtime/pkg/meta"
 	"github.com/pkg/errors"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -49,6 +50,7 @@ import (
 	"github.com/oam-dev/kubevela/pkg/oam/discoverymapper"
 	oamutil "github.com/oam-dev/kubevela/pkg/oam/util"
 	"github.com/oam-dev/kubevela/pkg/utils/apply"
+	"github.com/oam-dev/kubevela/pkg/workflow"
 	"github.com/oam-dev/kubevela/version"
 )
 
@@ -58,6 +60,8 @@ const (
 )
 
 const (
+	// WorkflowReconcileWaitTime is the time to wait before reconcile again workflow running
+	WorkflowReconcileWaitTime      = time.Second * 3
 	legacyResourceTrackerFinalizer = "resourceTracker.finalizer.core.oam.dev"
 	// resourceTrackerFinalizer is to delete the resource tracker of the latest app revision.
 	resourceTrackerFinalizer = "app.oam.dev/resource-tracker-finalizer"
@@ -149,6 +153,14 @@ func (r *Reconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
 		r.Recorder.Event(app, event.Warning(velatypes.ReasonFailedRender, err))
 		return handler.handleErr(err)
 	}
+	policies, wfSteps, err := generatedAppfile.GenerateWorkflowAndPolicy()
+	if err != nil {
+		klog.Error(err, "[Handle GenerateWorkflowAndPolicy]")
+		app.Status.SetConditions(errorCondition("Built", err))
+		r.Recorder.Event(app, event.Warning(velatypes.ReasonFailedRender, err))
+		return handler.handleErr(err)
+	}
+
 	app.Status.SetConditions(readyCondition("Built"))
 	r.Recorder.Event(app, event.Normal(velatypes.ReasonRendered, velatypes.MessageRendered))
 	klog.Info("Successfully render application resources", "application", klog.KObj(app))
@@ -156,7 +168,7 @@ func (r *Reconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
 	// pass application's labels and annotations to ac
 	oamutil.PassLabelAndAnnotation(app, ac)
 	// apply application resources' manifests to the cluster
-	if err := handler.apply(ctx, appRev, ac, comps); err != nil {
+	if err := handler.apply(ctx, appRev, ac, comps, policies); err != nil {
 		klog.ErrorS(err, "Failed to apply application resources' manifests",
 			"application", klog.KObj(app))
 		app.Status.SetConditions(errorCondition("Applied", err))
@@ -165,6 +177,17 @@ func (r *Reconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
 	}
 	klog.Info("Successfully apply application resources' manifests", "application", klog.KObj(app))
 
+	done, err := workflow.NewWorkflow(app, handler.r.applicator).ExecuteSteps(ctx, appRev.Name, wfSteps)
+	if err != nil {
+		klog.Error(err, "[handle workflow]")
+		app.Status.SetConditions(errorCondition("Workflow", err))
+		r.Recorder.Event(app, event.Warning(velatypes.ReasonFailedWorkflow, err))
+		return handler.handleErr(err)
+	}
+	if !done {
+		return reconcile.Result{RequeueAfter: WorkflowReconcileWaitTime}, r.UpdateStatus(ctx, app)
+	}
+
 	// if inplace is false and rolloutPlan is nil, it means the user will use an outer AppRollout object to rollout the application
 	if handler.app.Spec.RolloutPlan != nil {
 		res, err := handler.handleRollout(ctx)

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

@@ -26,8 +26,6 @@ import (
 	"strconv"
 	"time"
 
-	velatypes "github.com/oam-dev/kubevela/apis/types"
-
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 
@@ -49,6 +47,7 @@ import (
 	"github.com/oam-dev/kubevela/apis/core.oam.dev/common"
 	"github.com/oam-dev/kubevela/apis/core.oam.dev/v1alpha2"
 	"github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1"
+	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/util"
@@ -902,7 +901,7 @@ spec:
 	})
 
 	It("app-with-trait will create workload and trait with http task", func() {
-		s := NewMock()
+		s := newMockHTTP()
 		defer s.Close()
 		expTrait := expectScalerTrait(appWithTrait.Spec.Components[0].Name, appWithTrait.Name)
 		expTrait.Object["spec"].(map[string]interface{})["token"] = "test-token"
@@ -2334,7 +2333,7 @@ spec:
 `
 )
 
-func NewMock() *httptest.Server {
+func newMockHTTP() *httptest.Server {
 	ts := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		if r.Method != "GET" {
 			fmt.Printf("Expected 'GET' request, got '%s'", r.Method)

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

@@ -28,6 +28,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	ctypes "k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/klog/v2"
@@ -92,11 +93,15 @@ func (h *appHandler) handleErr(err error) (ctrl.Result, error) {
 	}, nil
 }
 
-func (h *appHandler) apply(ctx context.Context, appRev *v1beta1.ApplicationRevision, ac *v1alpha2.ApplicationConfiguration, comps []*v1alpha2.Component) error {
-	// don't create components if revision-only annotation is set
-	if ac.Annotations[oam.AnnotationAppRevisionOnly] == "true" {
+func (h *appHandler) apply(ctx context.Context, appRev *v1beta1.ApplicationRevision, ac *v1alpha2.ApplicationConfiguration, comps []*v1alpha2.Component, policies []*unstructured.Unstructured) error {
+
+	if h.app.Spec.Workflow != nil || ac.Annotations[oam.AnnotationAppRevisionOnly] == "true" {
 		h.FinalizeAppRevision(appRev, ac, comps)
-		return h.createOrUpdateAppRevision(ctx, appRev)
+		err := h.createOrUpdateAppRevision(ctx, appRev)
+		if err != nil {
+			return err
+		}
+		return h.createResourcesConfigMap(ctx, appRev, ac, comps, policies)
 	}
 
 	owners := []metav1.OwnerReference{{

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

@@ -17,13 +17,17 @@ limitations under the License.
 package application
 
 import (
+	"bytes"
 	"context"
 	"sort"
 
 	"github.com/pkg/errors"
+	corev1 "k8s.io/api/core/v1"
 	apiequality "k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/klog/v2"
 	"k8s.io/utils/pointer"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -85,6 +89,66 @@ func (h *appHandler) setRevisionWithRenderedResult(appRev *v1beta1.ApplicationRe
 	comps []*v1alpha2.Component) {
 	appRev.Spec.Components = ConvertComponents2RawRevisions(comps)
 	appRev.Spec.ApplicationConfiguration = util.Object2RawExtension(ac)
+	appRev.Spec.ResourcesConfigMap.Name = appRev.Name
+}
+
+const (
+	// ConfigMapKeyResources is the key in ConfigMap Data field for containing data of resources
+	ConfigMapKeyResources = "resources"
+)
+
+func (h *appHandler) createResourcesConfigMap(ctx context.Context,
+	appRev *v1beta1.ApplicationRevision,
+	ac *v1alpha2.ApplicationConfiguration,
+	comps []*v1alpha2.Component,
+	policies []*unstructured.Unstructured) error {
+
+	buf := &bytes.Buffer{}
+	for _, c := range comps {
+		r := makeFinalResource(c.Spec.Workload, c.Name, c.Namespace)
+		buf.Write(util.MustJSONMarshal(r))
+	}
+	for _, acc := range ac.Spec.Components {
+		for _, tr := range acc.Traits {
+			r := makeFinalResource(tr.Trait, acc.ComponentName, ac.Namespace)
+			buf.Write(util.MustJSONMarshal(r))
+		}
+	}
+	for _, policy := range policies {
+		buf.Write(util.MustJSONMarshal(policy))
+	}
+
+	cm := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      appRev.Name,
+			Namespace: appRev.Namespace,
+			OwnerReferences: []metav1.OwnerReference{
+				*metav1.NewControllerRef(appRev, v1beta1.ApplicationRevisionGroupVersionKind),
+			},
+		},
+		Data: map[string]string{
+			ConfigMapKeyResources: buf.String(),
+		},
+	}
+
+	err := h.r.Client.Get(ctx, client.ObjectKey{Name: appRev.Name, Namespace: appRev.Namespace}, &corev1.ConfigMap{})
+	if err == nil {
+		return nil
+	}
+	if err != nil && !apierrors.IsNotFound(err) {
+		return err
+	}
+	return h.r.Client.Create(ctx, cm)
+}
+
+func makeFinalResource(raw runtime.RawExtension, name, ns string) *unstructured.Unstructured {
+	u, err := util.Object2Unstructured(raw)
+	if err != nil {
+		panic(err)
+	}
+	u.SetName(name)
+	u.SetNamespace(ns)
+	return u
 }
 
 // gatherRevisionSpec will gather all revision spec withouth metadata and rendered result.
@@ -196,7 +260,6 @@ func (h *appHandler) FinalizeAppRevision(appRev *v1beta1.ApplicationRevision,
 
 	h.setRevisionMetadata(appRev)
 	h.setRevisionWithRenderedResult(appRev, ac, comps)
-
 }
 
 // ConvertComponents2RawRevisions convert to ComponentMap

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

@@ -230,7 +230,7 @@ var _ = Describe("test generate revision ", func() {
 		Expect(ac.Namespace).Should(Equal(app.Namespace))
 		appRev, err := handler.GenerateAppRevision(ctx)
 		Expect(err).Should(Succeed())
-		Expect(handler.apply(context.Background(), appRev, ac, comps)).Should(Succeed())
+		Expect(handler.apply(context.Background(), appRev, ac, comps, nil)).Should(Succeed())
 
 		curApp := &v1beta1.Application{}
 		Eventually(
@@ -265,7 +265,7 @@ var _ = Describe("test generate revision ", func() {
 		lastRevision := curApp.Status.LatestRevision.Name
 		appRev, err = handler.GenerateAppRevision(ctx)
 		Expect(err).Should(Succeed())
-		Expect(handler.apply(context.Background(), appRev, ac, comps)).Should(Succeed())
+		Expect(handler.apply(context.Background(), appRev, ac, comps, nil)).Should(Succeed())
 		Eventually(
 			func() error {
 				return handler.r.Get(ctx,
@@ -305,7 +305,7 @@ var _ = Describe("test generate revision ", func() {
 		handler.app = &app
 		appRev, err = handler.GenerateAppRevision(ctx)
 		Expect(err).Should(Succeed())
-		Expect(handler.apply(context.Background(), appRev, ac, comps)).Should(Succeed())
+		Expect(handler.apply(context.Background(), appRev, ac, comps, nil)).Should(Succeed())
 		Eventually(
 			func() error {
 				return handler.r.Get(ctx,
@@ -348,7 +348,7 @@ var _ = Describe("test generate revision ", func() {
 		Expect(ac.Namespace).Should(Equal(app.Namespace))
 		appRev, err := handler.GenerateAppRevision(ctx)
 		Expect(err).Should(Succeed())
-		Expect(handler.apply(context.Background(), appRev, ac, comps)).Should(Succeed())
+		Expect(handler.apply(context.Background(), appRev, ac, comps, nil)).Should(Succeed())
 		curApp := &v1beta1.Application{}
 		Eventually(
 			func() error {
@@ -381,7 +381,7 @@ var _ = Describe("test generate revision ", func() {
 		lastRevision := curApp.Status.LatestRevision.Name
 		appRev, err = handler.GenerateAppRevision(ctx)
 		Expect(err).Should(Succeed())
-		Expect(handler.apply(context.Background(), appRev, ac, comps)).Should(Succeed())
+		Expect(handler.apply(context.Background(), appRev, ac, comps, nil)).Should(Succeed())
 		Eventually(
 			func() error {
 				return handler.r.Get(ctx,
@@ -423,7 +423,7 @@ var _ = Describe("test generate revision ", func() {
 		handler.app = &app
 		appRev, err = handler.GenerateAppRevision(ctx)
 		Expect(err).Should(Succeed())
-		Expect(handler.apply(context.Background(), appRev, ac, comps)).Should(Succeed())
+		Expect(handler.apply(context.Background(), appRev, ac, comps, nil)).Should(Succeed())
 		Eventually(
 			func() error {
 				return handler.r.Get(ctx,
@@ -470,7 +470,7 @@ var _ = Describe("test generate revision ", func() {
 		Expect(ac.Namespace).Should(Equal(app.Namespace))
 		appRev, err := handler.GenerateAppRevision(ctx)
 		Expect(err).Should(Succeed())
-		Expect(handler.apply(context.Background(), appRev, ac, comps)).Should(Succeed())
+		Expect(handler.apply(context.Background(), appRev, ac, comps, nil)).Should(Succeed())
 
 		curApp := &v1beta1.Application{}
 		Eventually(
@@ -500,7 +500,7 @@ var _ = Describe("test generate revision ", func() {
 		lastRevision := curApp.Status.LatestRevision.Name
 		appRev, err = handler.GenerateAppRevision(ctx)
 		Expect(err).Should(Succeed())
-		Expect(handler.apply(context.Background(), appRev, ac, comps)).Should(Succeed())
+		Expect(handler.apply(context.Background(), appRev, ac, comps, nil)).Should(Succeed())
 		Eventually(
 			func() error {
 				return handler.r.Get(ctx, types.NamespacedName{Namespace: ns.Name, Name: app.Name}, curApp)

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

@@ -31,6 +31,7 @@ import (
 	. "github.com/onsi/gomega"
 	"github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
+	crdv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -51,6 +52,7 @@ import (
 	"github.com/oam-dev/kubevela/pkg/controller/core.oam.dev/v1alpha2/applicationconfiguration"
 	"github.com/oam-dev/kubevela/pkg/cue/packages"
 	"github.com/oam-dev/kubevela/pkg/oam/discoverymapper"
+	"github.com/oam-dev/kubevela/pkg/utils/apply"
 	// +kubebuilder:scaffold:imports
 )
 
@@ -120,6 +122,8 @@ var _ = BeforeSuite(func(done Done) {
 
 	terraformv1beta1.AddToScheme(testScheme)
 
+	crdv1.AddToScheme(testScheme)
+
 	// +kubebuilder:scaffold:scheme
 	k8sClient, err = client.New(cfg, client.Options{Scheme: testScheme})
 	Expect(err).ToNot(HaveOccurred())
@@ -135,6 +139,7 @@ var _ = BeforeSuite(func(done Done) {
 		pd:               pd,
 		Recorder:         event.NewAPIRecorder(recorder),
 		appRevisionLimit: appRevisionLimit,
+		applicator:       apply.NewAPIApplicator(k8sClient),
 	}
 	// setup the controller manager since we need the component handler to run in the background
 	ctlManager, err = ctrl.NewManager(cfg, ctrl.Options{

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

@@ -0,0 +1,361 @@
+/*
+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 application
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/ghodss/yaml"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	corev1 "k8s.io/api/core/v1"
+	crdv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+
+	oamcore "github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1"
+	"github.com/oam-dev/kubevela/pkg/oam/util"
+	"github.com/oam-dev/kubevela/pkg/workflow"
+)
+
+var _ = Describe("Test Workflow", func() {
+	ctx := context.Background()
+	namespace := "test-workflow"
+
+	appWithWorkflow := &oamcore.Application{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-app",
+			Namespace: namespace,
+		},
+		Spec: oamcore.ApplicationSpec{
+			Components: []oamcore.ApplicationComponent{{
+				Name:       "test-component",
+				Type:       "worker",
+				Properties: runtime.RawExtension{Raw: []byte(`{"cmd":["sleep","1000"],"image":"busybox"}`)},
+			}},
+			Workflow: []oamcore.WorkflowStep{{
+				Name:       "test-wf1",
+				Type:       "foowf",
+				Properties: runtime.RawExtension{Raw: []byte(`{"key":"test"}`)},
+			}, {
+				Name:       "test-wf2",
+				Type:       "foowf",
+				Properties: runtime.RawExtension{Raw: []byte(`{"key":"test"}`)},
+			}},
+		},
+	}
+	appWithWorkflowAndPolicy := appWithWorkflow.DeepCopy()
+	appWithWorkflowAndPolicy.Name = "test-wf-policy"
+	appWithWorkflowAndPolicy.Spec.Policies = []oamcore.AppPolicy{{
+		Name:       "test-policy",
+		Type:       "foopolicy",
+		Properties: runtime.RawExtension{Raw: []byte(`{"key":"test"}`)},
+	}}
+
+	testDefinitions := []string{componentDefYaml, policyDefYaml, wfStepDefYaml}
+
+	BeforeEach(func() {
+		setupFooCRD(ctx)
+		setupNamespace(ctx, namespace)
+		setupTestDefinitions(ctx, testDefinitions, namespace)
+		By("[TEST] Set up definitions before integration test")
+	})
+
+	AfterEach(func() {
+	})
+
+	It("should create ConfigMap with final resources for app with workflow", func() {
+		Expect(k8sClient.Create(ctx, appWithWorkflowAndPolicy)).Should(BeNil())
+
+		// first try to add finalizer
+		tryReconcile(reconciler, appWithWorkflowAndPolicy.Name, appWithWorkflowAndPolicy.Namespace)
+		tryReconcile(reconciler, appWithWorkflowAndPolicy.Name, appWithWorkflowAndPolicy.Namespace)
+
+		appRev := &oamcore.ApplicationRevision{}
+		Expect(k8sClient.Get(ctx, client.ObjectKey{
+			Name:      appWithWorkflowAndPolicy.Name + "-v1",
+			Namespace: namespace,
+		}, appRev)).Should(BeNil())
+
+		Expect(appRev.Spec.ResourcesConfigMap.Name).ShouldNot(BeEmpty())
+
+		cm := &corev1.ConfigMap{}
+		Expect(k8sClient.Get(ctx, client.ObjectKey{
+			Name:      appRev.Name,
+			Namespace: namespace,
+		}, cm)).Should(BeNil())
+
+		Expect(cm.Data["resources"]).Should(Equal(compressJSON(appWithWorkflowAndPolicyResources)))
+	})
+
+	It("should execute workflow steps one by one", func() {
+		Expect(k8sClient.Create(ctx, appWithWorkflow)).Should(BeNil())
+
+		// first try to add finalizer
+		tryReconcile(reconciler, appWithWorkflowAndPolicy.Name, appWithWorkflowAndPolicy.Namespace)
+		tryReconcile(reconciler, appWithWorkflow.Name, appWithWorkflow.Namespace)
+
+		// check step 1 created, step 2 not
+		step1obj := &unstructured.Unstructured{}
+		step1obj.SetGroupVersionKind(schema.GroupVersionKind{
+			Group:   "example.com",
+			Kind:    "Foo",
+			Version: "v1",
+		})
+
+		step2obj := &unstructured.Unstructured{}
+		step2obj.SetGroupVersionKind(schema.GroupVersionKind{
+			Group:   "example.com",
+			Kind:    "Foo",
+			Version: "v1",
+		})
+
+		Expect(k8sClient.Get(ctx, client.ObjectKey{
+			Name:      "test-wf1",
+			Namespace: appWithWorkflow.Namespace,
+		}, step1obj)).Should(BeNil())
+
+		Expect(k8sClient.Get(ctx, client.ObjectKey{
+			Name:      "test-wf2",
+			Namespace: appWithWorkflow.Namespace,
+		}, step2obj)).Should(&util.NotFoundMatcher{})
+
+		// mark step 1 succeeded, reconcile
+		markWorkflowSucceeded(step1obj)
+		Expect(k8sClient.Update(ctx, step1obj)).Should(BeNil())
+
+		tryReconcile(reconciler, appWithWorkflow.Name, appWithWorkflow.Namespace)
+		// check step 2 created
+		Expect(k8sClient.Get(ctx, client.ObjectKey{
+			Name:      "test-wf2",
+			Namespace: appWithWorkflow.Namespace,
+		}, step2obj)).Should(BeNil())
+	})
+})
+
+func markWorkflowSucceeded(obj *unstructured.Unstructured) {
+	succeededMessage, _ := json.Marshal(&workflow.SucceededMessage{ObservedGeneration: 2})
+
+	m := map[string]interface{}{
+		"conditions": []interface{}{
+			map[string]interface{}{
+				"type":    workflow.CondTypeWorkflowFinish,
+				"reason":  workflow.CondReasonSucceeded,
+				"message": string(succeededMessage),
+				"status":  workflow.CondStatusTrue,
+			},
+		},
+	}
+
+	unstructured.SetNestedMap(obj.Object, m, "status")
+}
+
+func compressJSON(d string) string {
+	m := &json.RawMessage{}
+	r := bytes.NewBuffer([]byte(d))
+	dec := json.NewDecoder(r)
+	w := &bytes.Buffer{}
+
+	for {
+		err := dec.Decode(m)
+		if err != nil {
+			break
+		}
+		b, _ := json.Marshal(m)
+		w.Write(b)
+	}
+	return w.String()
+}
+
+func tryReconcile(r *Reconciler, name, ns string) {
+	appKey := client.ObjectKey{
+		Name:      name,
+		Namespace: ns,
+	}
+
+	Eventually(func() error {
+		_, err := r.Reconcile(reconcile.Request{NamespacedName: appKey})
+		if err != nil {
+			By(fmt.Sprintf("reconcile err: %+v ", err))
+		}
+		return err
+	}, 10*time.Second, time.Second).Should(BeNil())
+}
+
+func setupNamespace(ctx context.Context, namespace string) {
+	ns := &corev1.Namespace{
+		ObjectMeta: metav1.ObjectMeta{Name: namespace},
+	}
+	Expect(k8sClient.Create(ctx, ns)).Should(SatisfyAny(BeNil(), &util.AlreadyExistMatcher{}))
+}
+
+func setupFooCRD(ctx context.Context) {
+	trueVar := true
+	foocrd := &crdv1.CustomResourceDefinition{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "foo.example.com",
+		},
+		Spec: crdv1.CustomResourceDefinitionSpec{
+			Group: "example.com",
+			Names: crdv1.CustomResourceDefinitionNames{
+				Kind:     "Foo",
+				ListKind: "FooList",
+				Plural:   "foo",
+				Singular: "foo",
+			},
+			Versions: []crdv1.CustomResourceDefinitionVersion{{
+				Name:    "v1",
+				Served:  true,
+				Storage: true,
+				Schema: &crdv1.CustomResourceValidation{
+					OpenAPIV3Schema: &crdv1.JSONSchemaProps{
+						Type: "object",
+						Properties: map[string]crdv1.JSONSchemaProps{
+							"spec": {
+								Type: "object",
+								Properties: map[string]crdv1.JSONSchemaProps{
+									"key": {Type: "string"},
+								},
+							},
+						},
+						XPreserveUnknownFields: &trueVar,
+					},
+				},
+			},
+			},
+			Scope: crdv1.NamespaceScoped,
+		},
+	}
+	Expect(k8sClient.Create(ctx, foocrd)).Should(SatisfyAny(BeNil(), &util.AlreadyExistMatcher{}))
+}
+
+func setupTestDefinitions(ctx context.Context, defs []string, ns string) {
+	for _, def := range defs {
+		defJson, err := yaml.YAMLToJSON([]byte(def))
+		Expect(err).Should(BeNil())
+		u := &unstructured.Unstructured{}
+		Expect(json.Unmarshal(defJson, u)).Should(BeNil())
+		u.SetNamespace(ns)
+		Expect(k8sClient.Create(ctx, u)).Should(SatisfyAny(BeNil(), &util.AlreadyExistMatcher{}))
+	}
+}
+
+const (
+	policyDefYaml = `apiVersion: core.oam.dev/v1beta1
+kind: PolicyDefinition
+metadata:
+  name: foopolicy
+spec:
+  schematic:
+    cue:
+      template: |
+        output: {
+          apiVersion: "example.com/v1"
+          kind:       "Foo"
+          spec: {
+            key: parameter.key
+          }
+        }
+        parameter: {
+          key: string
+        }
+`
+	wfStepDefYaml = `apiVersion: core.oam.dev/v1beta1
+kind: WorkflowStepDefinition
+metadata:
+  name: foowf
+spec:
+  schematic:
+    cue:
+      template: |
+        output: {
+          apiVersion: "example.com/v1"
+          kind:       "Foo"
+          spec: {
+            key: parameter.key
+          }
+        }
+        parameter: {
+          key: string
+        }
+`
+
+	appWithWorkflowAndPolicyResources = `{
+  "apiVersion": "apps/v1",
+  "kind": "Deployment",
+  "metadata": {
+    "annotations": {},
+    "labels": {
+      "app.oam.dev/appRevision": "test-wf-policy-v1",
+      "app.oam.dev/component": "test-component",
+      "app.oam.dev/name": "test-wf-policy",
+      "workload.oam.dev/type": "worker"
+    },
+    "name": "test-component",
+    "namespace": "test-workflow"
+  },
+  "spec": {
+    "selector": {
+      "matchLabels": {
+        "app.oam.dev/component": "test-component"
+      }
+    },
+    "template": {
+      "metadata": {
+        "labels": {
+          "app.oam.dev/component": "test-component"
+        }
+      },
+      "spec": {
+        "containers": [
+          {
+            "command": [
+              "sleep",
+              "1000"
+            ],
+            "image": "busybox",
+            "name": "test-component"
+          }
+        ]
+      }
+    }
+  }
+}
+{
+  "apiVersion": "example.com/v1",
+  "kind": "Foo",
+  "metadata": {
+    "labels": {
+      "app.oam.dev/appRevision": "test-wf-policy-v1",
+      "app.oam.dev/component": "test-policy",
+      "app.oam.dev/name": "test-wf-policy",
+      "workload.oam.dev/type": "foopolicy"
+    }
+  },
+  "spec": {
+    "key": "test"
+  }
+}`
+)