From e5c8f259f30e02c196f2d5e5faf0a4dbb67ca129 Mon Sep 17 00:00:00 2001 From: Hongchao Deng Date: Mon, 14 Jun 2021 22:40:12 -0700 Subject: [PATCH] app controller: reconcile workflow and create res configmap (#1788) --- docs/examples/workflow/README.md | 130 +++++++ docs/examples/workflow/app.yaml | 28 ++ docs/examples/workflow/definition.yaml | 52 +++ docs/examples/workflow/wf-patch.yaml | 6 + pkg/appfile/appfile.go | 36 ++ pkg/appfile/parser.go | 37 ++ .../application/application_controller.go | 25 +- .../application_controller_test.go | 7 +- .../v1alpha2/application/apply.go | 13 +- .../v1alpha2/application/revision.go | 65 +++- .../v1alpha2/application/revision_test.go | 16 +- .../v1alpha2/application/suite_test.go | 5 + .../v1alpha2/application/workflow_test.go | 361 ++++++++++++++++++ 13 files changed, 763 insertions(+), 18 deletions(-) create mode 100644 docs/examples/workflow/README.md create mode 100644 docs/examples/workflow/app.yaml create mode 100644 docs/examples/workflow/definition.yaml create mode 100644 docs/examples/workflow/wf-patch.yaml create mode 100644 pkg/controller/core.oam.dev/v1alpha2/application/workflow_test.go diff --git a/docs/examples/workflow/README.md b/docs/examples/workflow/README.md new file mode 100644 index 000000000..2199decdb --- /dev/null +++ b/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 + ``` \ No newline at end of file diff --git a/docs/examples/workflow/app.yaml b/docs/examples/workflow/app.yaml new file mode 100644 index 000000000..7732c957d --- /dev/null +++ b/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 \ No newline at end of file diff --git a/docs/examples/workflow/definition.yaml b/docs/examples/workflow/definition.yaml new file mode 100644 index 000000000..e69df2441 --- /dev/null +++ b/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 + } \ No newline at end of file diff --git a/docs/examples/workflow/wf-patch.yaml b/docs/examples/workflow/wf-patch.yaml new file mode 100644 index 000000000..97116aa54 --- /dev/null +++ b/docs/examples/workflow/wf-patch.yaml @@ -0,0 +1,6 @@ +status: + conditions: + - type: 'workflow-finish' + status: 'True' + reason: 'Succeeded' + message: '{"observedGeneration":3}' \ No newline at end of file diff --git a/pkg/appfile/appfile.go b/pkg/appfile/appfile.go index 1426e0cb2..4e0dff612 100644 --- a/pkg/appfile/appfile.go +++ b/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 diff --git a/pkg/appfile/parser.go b/pkg/appfile/parser.go index ccc4d49ca..0f3fe643f 100644 --- a/pkg/appfile/parser.go +++ b/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) { 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 e8878fc18..8dd448bee 100644 --- a/pkg/controller/core.oam.dev/v1alpha2/application/application_controller.go +++ b/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) 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 e421632d1..b92e9fb26 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 @@ -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) diff --git a/pkg/controller/core.oam.dev/v1alpha2/application/apply.go b/pkg/controller/core.oam.dev/v1alpha2/application/apply.go index 73d762447..708c933f3 100644 --- a/pkg/controller/core.oam.dev/v1alpha2/application/apply.go +++ b/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{{ diff --git a/pkg/controller/core.oam.dev/v1alpha2/application/revision.go b/pkg/controller/core.oam.dev/v1alpha2/application/revision.go index 0daa5ccb0..151dd90bb 100644 --- a/pkg/controller/core.oam.dev/v1alpha2/application/revision.go +++ b/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 diff --git a/pkg/controller/core.oam.dev/v1alpha2/application/revision_test.go b/pkg/controller/core.oam.dev/v1alpha2/application/revision_test.go index 370add8be..b05af1916 100644 --- a/pkg/controller/core.oam.dev/v1alpha2/application/revision_test.go +++ b/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) diff --git a/pkg/controller/core.oam.dev/v1alpha2/application/suite_test.go b/pkg/controller/core.oam.dev/v1alpha2/application/suite_test.go index 1ed0914e8..44c61ce9b 100644 --- a/pkg/controller/core.oam.dev/v1alpha2/application/suite_test.go +++ b/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{ diff --git a/pkg/controller/core.oam.dev/v1alpha2/application/workflow_test.go b/pkg/controller/core.oam.dev/v1alpha2/application/workflow_test.go new file mode 100644 index 000000000..2692eec01 --- /dev/null +++ b/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" + } +}` +)