Feat: dispatch manifests in concurrent (#3060)

* Feat: dispatch manifests in concurrent

Signed-off-by: yangsoon <songyang.song@alibaba-inc.com>

* Fix: merge workflow pkg convert to pkg util

Signed-off-by: yangsoon <songyang.song@alibaba-inc.com>

Co-authored-by: yangsoon <songyang.song@alibaba-inc.com>

cmd/core/main.go

@@ -46,6 +46,7 @@ import (
 	"github.com/oam-dev/kubevela/pkg/multicluster"
 	"github.com/oam-dev/kubevela/pkg/oam"
 	"github.com/oam-dev/kubevela/pkg/oam/discoverymapper"
+	"github.com/oam-dev/kubevela/pkg/resourcekeeper"
 	"github.com/oam-dev/kubevela/pkg/utils/common"
 	"github.com/oam-dev/kubevela/pkg/utils/system"
 	oamwebhook "github.com/oam-dev/kubevela/pkg/webhook/core.oam.dev"
@@ -132,6 +133,7 @@ func main() {
 	flag.BoolVar(&controllerArgs.EnableCompatibility, "enable-asi-compatibility", false, "enable compatibility for asi")
 	flag.BoolVar(&controllerArgs.IgnoreAppWithoutControllerRequirement, "ignore-app-without-controller-version", false, "If true, application controller will not process the app without 'app.oam.dev/controller-version-require' annotation")
 	standardcontroller.AddOptimizeFlags()
+	flag.IntVar(&resourcekeeper.MaxDispatchConcurrent, "max-dispatch-concurrent", 10, "Set the max dispatch concurrent number, default is 10")
 
 	flag.Parse()
 	// setup logging

pkg/client/controller_client.go

@@ -21,7 +21,6 @@ import (
 	"strings"
 	"sync"
 
-	"github.com/oam-dev/kubevela/pkg/oam"
 	"github.com/oam-dev/kubevela/pkg/resourcetracker"
 
 	"github.com/pkg/errors"
@@ -87,15 +86,13 @@ func DefaultNewControllerClient(cache cache.Cache, config *rest.Config, options
 		AddFunc: func(obj interface{}) {
 			lock.Lock()
 			rtCount++
-			metrics.ResourceTrackerNumberGauge.WithLabelValues(
-				metrics.ExtractMetricValuesFromObjectLabel(obj, oam.LabelAppName, oam.LabelAppNamespace)...).Set(float64(rtCount))
+			metrics.ResourceTrackerNumberGauge.WithLabelValues("application").Set(float64(rtCount))
 			lock.Unlock()
 		},
 		DeleteFunc: func(obj interface{}) {
 			lock.Lock()
 			rtCount--
-			metrics.ResourceTrackerNumberGauge.WithLabelValues(
-				metrics.ExtractMetricValuesFromObjectLabel(obj, oam.LabelAppName, oam.LabelAppNamespace)...).Set(float64(rtCount))
+			metrics.ResourceTrackerNumberGauge.WithLabelValues("application").Set(float64(rtCount))
 			lock.Unlock()
 		},
 	})

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

@@ -2275,7 +2275,51 @@ var _ = Describe("Test Application Controller", func() {
 			Namespace: app.Namespace,
 		}, checkWeb)).Should(BeNil())
 		Expect(*(checkWeb.Spec.Replicas)).Should(BeEquivalentTo(int32(0)))
+	})
 
+	It("app apply resource in parallel", func() {
+		wfDef := &v1beta1.WorkflowStepDefinition{}
+		wfDefJson, _ := yaml.YAMLToJSON([]byte(applyInParallelWorkflowDefinitionYaml))
+		Expect(json.Unmarshal(wfDefJson, wfDef)).Should(BeNil())
+		Expect(k8sClient.Create(ctx, wfDef.DeepCopy())).Should(SatisfyAny(BeNil(), &util.AlreadyExistMatcher{}))
+
+		ns := &corev1.Namespace{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "vela-test-apply-in-parallel",
+			},
+		}
+		app := appwithNoTrait.DeepCopy()
+		app.Name = "vela-test-app"
+		app.SetNamespace(ns.Name)
+		app.Spec.Workflow = &v1beta1.Workflow{
+			Steps: []v1beta1.WorkflowStep{{
+				Name:       "apply-in-parallel",
+				Type:       "apply-test",
+				Properties: &runtime.RawExtension{Raw: []byte(`{"parallelism": 20}`)},
+			}},
+		}
+		Expect(k8sClient.Create(ctx, ns)).Should(BeNil())
+		Expect(k8sClient.Create(ctx, app)).Should(BeNil())
+		appKey := client.ObjectKey{
+			Name:      app.Name,
+			Namespace: app.Namespace,
+		}
+		_, err := testutil.ReconcileOnceAfterFinalizer(reconciler, reconcile.Request{NamespacedName: appKey})
+		Expect(err).Should(BeNil())
+
+		deployList := new(v1.DeploymentList)
+		Expect(k8sClient.List(ctx, deployList, client.InNamespace(app.Namespace))).Should(BeNil())
+		Expect(len(deployList.Items)).Should(Equal(20))
+
+		checkApp := new(v1beta1.Application)
+		Expect(k8sClient.Get(ctx, appKey, checkApp)).Should(Succeed())
+		rt := new(v1beta1.ResourceTracker)
+		expectRTName := fmt.Sprintf("%s-%s", checkApp.Status.LatestRevision.Name, checkApp.GetNamespace())
+		Eventually(func() error {
+			return k8sClient.Get(ctx, client.ObjectKey{Name: expectRTName}, rt)
+		}, 10*time.Second, 500*time.Millisecond).Should(Succeed())
+
+		Expect(len(rt.Spec.ManagedResources)).Should(Equal(20))
 	})
 
 	It("test controller requirement", func() {
@@ -3252,6 +3296,46 @@ spec:
           }
         }
         parameter: objects: [...{}]
+`
+	applyInParallelWorkflowDefinitionYaml = `
+apiVersion: core.oam.dev/v1beta1
+kind: WorkflowStepDefinition
+metadata:
+  name: apply-test
+  namespace: vela-system
+spec:
+  schematic:
+    cue:
+      template: |
+        import (
+                "vela/op"
+                "list"
+        )
+
+        components:      op.#LoadInOrder & {}
+        targetComponent: components.value[0]
+        resources:       op.#RenderComponent & {
+                value: targetComponent
+        }
+        workload:       resources.output
+        arr:            list.Range(0, parameter.parallelism, 1)
+        patchWorkloads: op.#Steps & {
+                for idx in arr {
+                        "\(idx)": op.#PatchK8sObject & {
+                                value: workload
+                                patch: {
+                                        // +patchStrategy=retainKeys
+                                        metadata: name: "\(targetComponent.name)-\(idx)"
+                                }
+                        }
+                }
+        }
+        workloads: [ for patchResult in patchWorkloads {patchResult.result}]
+        apply: op.#ApplyInParallel & {
+                value: workloads
+        }
+        parameter: parallelism: int
+
 `
 )
 

pkg/monitor/metrics/application.go

@@ -18,7 +18,6 @@ package metrics
 
 import (
 	"github.com/prometheus/client_golang/prometheus"
-	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 var (
@@ -116,17 +115,5 @@ var (
 	ResourceTrackerNumberGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
 		Name: "resourcetracker_number",
 		Help: "resourceTracker number.",
-	}, []string{"application", "namespace"})
+	}, []string{"controller"})
 )
-
-// ExtractMetricValuesFromObjectLabel extract metric values from k8s object's labels
-func ExtractMetricValuesFromObjectLabel(obj interface{}, labelKeys ...string) (values []string) {
-	if resource, ok := obj.(client.Object); ok {
-		for _, labelKey := range labelKeys {
-			values = append(values, resource.GetLabels()[labelKey])
-		}
-	} else {
-		values = make([]string, len(labelKeys))
-	}
-	return
-}

pkg/resourcekeeper/componentrevision.go

@@ -39,7 +39,7 @@ func (h *resourceKeeper) DispatchComponentRevision(ctx context.Context, cr *v1.C
 	obj.SetName(cr.Name)
 	obj.SetNamespace(cr.Namespace)
 	obj.SetLabels(cr.Labels)
-	if err = resourcetracker.RecordManifestInResourceTracker(multicluster.ContextInLocalCluster(ctx), h.Client, rt, obj, true); err != nil {
+	if err = resourcetracker.RecordManifestsInResourceTracker(multicluster.ContextInLocalCluster(ctx), h.Client, rt, []*unstructured.Unstructured{obj}, true); err != nil {
 		return errors.Wrapf(err, "failed to record componentrevision %s/%s/%s", oam.GetCluster(cr), cr.Namespace, cr.Name)
 	}
 	if err = h.Client.Create(multicluster.ContextWithClusterName(ctx, oam.GetCluster(cr)), cr); err != nil {

pkg/resourcekeeper/dispatch.go

@@ -18,17 +18,21 @@ package resourcekeeper
 
 import (
 	"context"
+	"sync"
 
 	"github.com/pkg/errors"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	kerrors "k8s.io/apimachinery/pkg/util/errors"
 
-	"github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1"
 	"github.com/oam-dev/kubevela/pkg/multicluster"
 	"github.com/oam-dev/kubevela/pkg/oam"
 	"github.com/oam-dev/kubevela/pkg/resourcetracker"
 	"github.com/oam-dev/kubevela/pkg/utils/apply"
 )
 
+// MaxDispatchConcurrent is the max dispatch concurrent number
+var MaxDispatchConcurrent = 10
+
 // DispatchOption option for dispatch
 type DispatchOption interface {
 	ApplyToDispatchConfig(*dispatchConfig)
@@ -52,6 +56,21 @@ func (h *resourceKeeper) Dispatch(ctx context.Context, manifests []*unstructured
 	if h.applyOncePolicy != nil && h.applyOncePolicy.Enable {
 		options = append(options, MetaOnlyOption{})
 	}
+	// 1. record manifests in resourcetracker
+	if err = h.record(ctx, manifests, options...); err != nil {
+		return err
+	}
+	// 2. apply manifests
+	if err = h.dispatch(ctx, manifests); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (h *resourceKeeper) record(ctx context.Context, manifests []*unstructured.Unstructured, options ...DispatchOption) error {
+	var rootManifests []*unstructured.Unstructured
+	var versionManifests []*unstructured.Unstructured
+
 	for _, manifest := range manifests {
 		if manifest != nil {
 			_options := options
@@ -61,34 +80,61 @@ func (h *resourceKeeper) Dispatch(ctx context.Context, manifests []*unstructured
 				}
 			}
 			cfg := newDispatchConfig(_options...)
-			if err = h.dispatch(ctx, manifest, cfg); err != nil {
-				return err
+			if !cfg.skipRT {
+				if cfg.useRoot {
+					rootManifests = append(rootManifests, manifest)
+				} else {
+					versionManifests = append(versionManifests, manifest)
+				}
 			}
 		}
 	}
+
+	cfg := newDispatchConfig(options...)
+	if len(rootManifests) != 0 {
+		rt, err := h.getRootRT(ctx)
+		if err != nil {
+			return errors.Wrapf(err, "failed to get resourcetracker")
+		}
+		if err = resourcetracker.RecordManifestsInResourceTracker(multicluster.ContextInLocalCluster(ctx), h.Client, rt, rootManifests, cfg.metaOnly); err != nil {
+			return errors.Wrapf(err, "failed to record resources in resourcetracker %s", rt.Name)
+		}
+	}
+
+	rt, err := h.getCurrentRT(ctx)
+	if err != nil {
+		return errors.Wrapf(err, "failed to get resourcetracker")
+	}
+	if err = resourcetracker.RecordManifestsInResourceTracker(multicluster.ContextInLocalCluster(ctx), h.Client, rt, versionManifests, cfg.metaOnly); err != nil {
+		return errors.Wrapf(err, "failed to record resources in resourcetracker %s", rt.Name)
+	}
 	return nil
 }
 
-func (h *resourceKeeper) dispatch(ctx context.Context, manifest *unstructured.Unstructured, cfg *dispatchConfig) (err error) {
-	// 1. record manifests in resourcetracker
-	if !cfg.skipRT {
-		var rt *v1beta1.ResourceTracker
-		if cfg.useRoot {
-			rt, err = h.getRootRT(ctx)
-		} else {
-			rt, err = h.getCurrentRT(ctx)
-		}
-		if err != nil {
-			return errors.Wrapf(err, "failed to get resourcetracker")
-		}
-		if err = resourcetracker.RecordManifestInResourceTracker(multicluster.ContextInLocalCluster(ctx), h.Client, rt, manifest, cfg.metaOnly); err != nil {
-			return errors.Wrapf(err, "failed to record resources in resourcetracker %s", rt.Name)
-		}
-	}
-	// 2. apply manifests
+func (h *resourceKeeper) dispatch(ctx context.Context, manifests []*unstructured.Unstructured) error {
+	var errs []error
+	var l sync.Mutex
+	var wg sync.WaitGroup
+
+	ch := make(chan struct{}, MaxDispatchConcurrent)
 	applyOpts := []apply.ApplyOption{apply.MustBeControlledByApp(h.app), apply.NotUpdateRenderHashEqual()}
-	if err := h.applicator.Apply(multicluster.ContextWithClusterName(ctx, oam.GetCluster(manifest)), manifest, applyOpts...); err != nil {
-		return errors.Wrapf(err, "cannot apply manifest, name: %s apiVersion: %s kind: %s", manifest.GetName(), manifest.GetAPIVersion(), manifest.GetKind())
+
+	for i := 0; i < len(manifests); i++ {
+		ch <- struct{}{}
+		wg.Add(1)
+		go func(index int) {
+			defer wg.Done()
+			manifest := manifests[index]
+			applyCtx := multicluster.ContextWithClusterName(ctx, oam.GetCluster(manifest))
+			err := h.applicator.Apply(applyCtx, manifest, applyOpts...)
+			if err != nil {
+				l.Lock()
+				errs = append(errs, err)
+				l.Unlock()
+			}
+			<-ch
+		}(i)
 	}
-	return nil
+	wg.Wait()
+	return kerrors.NewAggregate(errs)
 }

pkg/resourcekeeper/gc_test.go

@@ -101,9 +101,9 @@ func TestResourceKeeperGarbageCollect(t *testing.T) {
 			obj.SetName(cr.GetName())
 			obj.SetNamespace(cr.GetNamespace())
 			obj.SetLabels(cr.GetLabels())
-			r.NoError(resourcetracker.RecordManifestInResourceTracker(ctx, cli, crRT, obj, true))
+			r.NoError(resourcetracker.RecordManifestsInResourceTracker(ctx, cli, crRT, []*unstructured.Unstructured{obj}, true))
 		}
-		r.NoError(resourcetracker.RecordManifestInResourceTracker(ctx, cli, _rt, cmMaps[i], true))
+		r.NoError(resourcetracker.RecordManifestsInResourceTracker(ctx, cli, _rt, []*unstructured.Unstructured{cmMaps[i]}, true))
 	}
 
 	checkCount := func(cmCount, rtCount int, crCount int) {

pkg/resourcetracker/app.go

@@ -142,12 +142,15 @@ func ListApplicationResourceTrackers(ctx context.Context, cli client.Client, app
 	return rootRT, currentRT, historyRTs, crRT, nil
 }
 
-// RecordManifestInResourceTracker records resources in ResourceTracker
-func RecordManifestInResourceTracker(ctx context.Context, cli client.Client, rt *v1beta1.ResourceTracker, manifest *unstructured.Unstructured, metaOnly bool) error {
-	if updated := rt.AddManagedResource(manifest, metaOnly); !updated {
-		return nil
+// RecordManifestsInResourceTracker records resources in ResourceTracker
+func RecordManifestsInResourceTracker(ctx context.Context, cli client.Client, rt *v1beta1.ResourceTracker, manifests []*unstructured.Unstructured, metaOnly bool) error {
+	if len(manifests) != 0 {
+		for _, manifest := range manifests {
+			rt.AddManagedResource(manifest, metaOnly)
+		}
+		return cli.Update(ctx, rt)
 	}
-	return cli.Update(ctx, rt)
+	return nil
 }
 
 // DeletedManifestInResourceTracker marks resources as deleted in resourcetracker, if remove is true, resources will be removed from resourcetracker

pkg/resourcetracker/app_test.go

@@ -97,7 +97,7 @@ func TestRecordAndDeleteManifestsInResourceTracker(t *testing.T) {
 		obj := &unstructured.Unstructured{}
 		obj.SetName(fmt.Sprintf("workload-%d", i))
 		objs = append(objs, obj)
-		r.NoError(RecordManifestInResourceTracker(context.Background(), cli, rt, obj, rand.Int()%2 == 0))
+		r.NoError(RecordManifestsInResourceTracker(context.Background(), cli, rt, []*unstructured.Unstructured{obj}, rand.Int()%2 == 0))
 	}
 	rand.Shuffle(len(objs), func(i, j int) { objs[i], objs[j] = objs[j], objs[i] })
 	for i := 0; i < n; i++ {

pkg/stdlib/op.cue

@@ -17,6 +17,8 @@ import (
 
 #Apply: kube.#Apply
 
+#ApplyInParallel: kube.#ApplyInParallel
+
 #Read: kube.#Read
 
 #List: kube.#List
@@ -156,7 +158,7 @@ import (
 
 #HTTPDelete: http.#Do & {method: "DELETE"}
 
-#ConvertString: convert.#String
+#ConvertString: util.#String
 
 #DateToTimestamp: time.#DateToTimestamp
 
@@ -168,6 +170,8 @@ import (
 
 #LoadInOrder: oam.#LoadComponetsInOrder
 
+#PatchK8sObject: util.#PatchK8sObject
+
 #Steps: {
 	#do: "steps"
 	...

pkg/stdlib/pkgs/convert.cue

@@ -1,8 +0,0 @@
-#String: {
-	#do:       "string"
-	#provider: "convert"
-
-	bt:   bytes
-	str?: string
-	...
-}

pkg/stdlib/pkgs/kube.cue

@@ -6,6 +6,14 @@
 	...
 }
 
+#ApplyInParallel: {
+	#do:       "apply-in-parallel"
+	#provider: "kube"
+	cluster:   *"" | string
+	value: [...{...}]
+	...
+}
+
 #Read: {
 	#do:       "read"
 	#provider: "kube"

pkg/stdlib/pkgs/util.cue

@@ -0,0 +1,17 @@
+#PatchK8sObject: {
+	#do:       "patch-k8s-object"
+	#provider: "util"
+	value: {...}
+	patch: {...}
+	result: {...}
+	...
+}
+
+#String: {
+	#do:       "string"
+	#provider: "util"
+
+	bt:   bytes
+	str?: string
+	...
+}

pkg/workflow/providers/convert/string.go

@@ -1,53 +0,0 @@
-/*
-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 convert
-
-import (
-	"github.com/oam-dev/kubevela/pkg/cue/model/value"
-	wfContext "github.com/oam-dev/kubevela/pkg/workflow/context"
-	"github.com/oam-dev/kubevela/pkg/workflow/providers"
-	"github.com/oam-dev/kubevela/pkg/workflow/types"
-)
-
-const (
-	// ProviderName is provider name for install.
-	ProviderName = "convert"
-)
-
-type provider struct {
-}
-
-// String convert byte to string
-func (h *provider) String(ctx wfContext.Context, v *value.Value, act types.Action) error {
-	b, err := v.LookupValue("bt")
-	if err != nil {
-		return err
-	}
-	s, err := b.CueValue().Bytes()
-	if err != nil {
-		return err
-	}
-	return v.FillObject(string(s), "str")
-}
-
-// Install register handlers to provider discover.
-func Install(p providers.Providers) {
-	prd := &provider{}
-	p.Register(ProviderName, map[string]providers.Handler{
-		"string": prd.String,
-	})
-}

pkg/workflow/providers/convert/string_test.go

@@ -1,73 +0,0 @@
-/*
-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 convert
-
-import (
-	"errors"
-	"testing"
-
-	"github.com/stretchr/testify/require"
-
-	"github.com/oam-dev/kubevela/pkg/cue/model/value"
-	"github.com/oam-dev/kubevela/pkg/workflow/providers"
-)
-
-func TestConvertString(t *testing.T) {
-	testCases := map[string]struct {
-		from        string
-		expected    string
-		expectedErr error
-	}{
-		"success": {
-			from:     `bt: 'test'`,
-			expected: "test",
-		},
-		"fail": {
-			from:        `bt: 123`,
-			expectedErr: errors.New("bt: cannot use value 123 (type int) as string|bytes"),
-		},
-	}
-
-	for name, tc := range testCases {
-		t.Run(name, func(t *testing.T) {
-			r := require.New(t)
-			v, err := value.NewValue(tc.from, nil, "")
-			r.NoError(err)
-			prd := &provider{}
-			err = prd.String(nil, v, nil)
-			if tc.expectedErr != nil {
-				r.Equal(tc.expectedErr.Error(), err.Error())
-				return
-			}
-			r.NoError(err)
-			expected, err := v.LookupValue("str")
-			r.NoError(err)
-			ret, err := expected.CueValue().String()
-			r.NoError(err)
-			r.Equal(ret, tc.expected)
-		})
-	}
-}
-
-func TestInstall(t *testing.T) {
-	p := providers.NewProviders()
-	Install(p)
-	h, ok := p.GetHandler("convert", "string")
-	r := require.New(t)
-	r.Equal(ok, true)
-	r.Equal(h != nil, true)
-}

pkg/workflow/providers/kube/handle.go

@@ -91,6 +91,40 @@ func (h *provider) Apply(ctx wfContext.Context, v *value.Value, act types.Action
 	return v.FillObject(workload.Object, "value")
 }
 
+// ApplyInParallel create or update CRs in parallel.
+func (h *provider) ApplyInParallel(ctx wfContext.Context, v *value.Value, act types.Action) error {
+	val, err := v.LookupValue("value")
+	if err != nil {
+		return err
+	}
+	iter, err := val.CueValue().List()
+	if err != nil {
+		return err
+	}
+	workloadNum := 0
+	for iter.Next() {
+		workloadNum++
+	}
+	var workloads = make([]*unstructured.Unstructured, workloadNum)
+	if err = val.UnmarshalTo(&workloads); err != nil {
+		return err
+	}
+	for i := range workloads {
+		if workloads[i].GetNamespace() == "" {
+			workloads[i].SetNamespace("default")
+		}
+	}
+	cluster, err := v.GetString("cluster")
+	if err != nil {
+		return err
+	}
+	deployCtx := multicluster.ContextWithClusterName(context.Background(), cluster)
+	if err = h.apply(deployCtx, cluster, common.WorkflowResourceCreator, workloads...); err != nil {
+		return v.FillObject(err, "err")
+	}
+	return nil
+}
+
 // Read get CR from cluster.
 func (h *provider) Read(ctx wfContext.Context, v *value.Value, act types.Action) error {
 	val, err := v.LookupValue("value")
@@ -188,9 +222,10 @@ func Install(p providers.Providers, cli client.Client, apply Dispatcher, deleter
 		cli:    cli,
 	}
 	p.Register(ProviderName, map[string]providers.Handler{
-		"apply":  prd.Apply,
-		"read":   prd.Read,
-		"list":   prd.List,
-		"delete": prd.Delete,
+		"apply":             prd.Apply,
+		"apply-in-parallel": prd.ApplyInParallel,
+		"read":              prd.Read,
+		"list":              prd.List,
+		"delete":            prd.Delete,
 	})
 }

pkg/workflow/providers/kube/handle_test.go

@@ -158,6 +158,7 @@ cluster: ""
 		err = result.FillObject(expected.Object)
 		Expect(err).ToNot(HaveOccurred())
 	})
+
 	It("patch & apply", func() {
 		p := &provider{
 			apply: func(ctx context.Context, _ string, _ common.ResourceCreatorRole, manifests ...*unstructured.Unstructured) error {
@@ -409,7 +410,6 @@ val: {
 		err = p.Apply(ctx, v, nil)
 		Expect(err).To(HaveOccurred())
 	})
-
 })
 
 func newWorkflowContextForTest() (wfContext.Context, error) {

pkg/workflow/providers/oam/apply.go

@@ -20,13 +20,12 @@ import (
 	"encoding/json"
 	"strings"
 
-	"github.com/oam-dev/kubevela/pkg/cue/model/sets"
-
 	"github.com/pkg/errors"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 
 	"github.com/oam-dev/kubevela/apis/core.oam.dev/common"
 	"github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1"
+	"github.com/oam-dev/kubevela/pkg/cue/model/sets"
 	"github.com/oam-dev/kubevela/pkg/cue/model/value"
 	"github.com/oam-dev/kubevela/pkg/oam"
 	wfContext "github.com/oam-dev/kubevela/pkg/workflow/context"

pkg/workflow/providers/util/util.go

@@ -0,0 +1,82 @@
+/*
+ Copyright 2022. 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 util
+
+import (
+	"github.com/oam-dev/kubevela/pkg/cue/model"
+	"github.com/oam-dev/kubevela/pkg/cue/model/value"
+	wfContext "github.com/oam-dev/kubevela/pkg/workflow/context"
+	"github.com/oam-dev/kubevela/pkg/workflow/providers"
+	"github.com/oam-dev/kubevela/pkg/workflow/types"
+)
+
+const (
+	// ProviderName is provider name for install.
+	ProviderName = "util"
+)
+
+type provider struct{}
+
+func (p *provider) PatchK8sObject(ctx wfContext.Context, v *value.Value, act types.Action) error {
+	val, err := v.LookupValue("value")
+	if err != nil {
+		return err
+	}
+	pv, err := v.LookupValue("patch")
+	if err != nil {
+		return err
+	}
+	base, err := model.NewBase(val.CueValue())
+	if err != nil {
+		return err
+	}
+	patcher, err := model.NewOther(pv.CueValue())
+	if err != nil {
+		return err
+	}
+	if err = base.Unify(patcher); err != nil {
+		return v.FillObject(err, "err")
+	}
+
+	workload, err := base.Unstructured()
+	if err != nil {
+		return v.FillObject(err, "err")
+	}
+	return v.FillObject(workload.Object, "result")
+}
+
+// String convert byte to string
+func (p *provider) String(ctx wfContext.Context, v *value.Value, act types.Action) error {
+	b, err := v.LookupValue("bt")
+	if err != nil {
+		return err
+	}
+	s, err := b.CueValue().Bytes()
+	if err != nil {
+		return err
+	}
+	return v.FillObject(string(s), "str")
+}
+
+// Install register handlers to provider discover.
+func Install(p providers.Providers) {
+	prd := &provider{}
+	p.Register(ProviderName, map[string]providers.Handler{
+		"patch-k8s-object": prd.PatchK8sObject,
+		"string":           prd.String,
+	})
+}

pkg/workflow/providers/util/util_test.go

@@ -0,0 +1,202 @@
+/*
+ Copyright 2022. 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 util
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/oam-dev/kubevela/pkg/cue/model/value"
+	"github.com/oam-dev/kubevela/pkg/workflow/providers"
+)
+
+func TestPatchK8sObject(t *testing.T) {
+	testcases := map[string]struct {
+		value       string
+		expectedErr error
+		patchResult string
+	}{
+		"test patch k8s object": {
+			value: `
+value: {
+	apiVersion: "apps/v1"
+	kind:       "Deployment"
+	spec: template: metadata: {
+		labels: {
+			"oam.dev/name": "test"
+		}
+	}
+}
+patch: {
+	spec: template: metadata: {
+		labels: {
+			"test-label": "true"
+		}
+	}
+}
+`,
+			expectedErr: nil,
+			patchResult: `
+apiVersion: "apps/v1"
+kind:       "Deployment"
+spec: template: metadata: {
+	labels: {
+		"oam.dev/name": "test"
+		"test-label":   "true"
+	}
+}
+`,
+		},
+		"test patch k8s object with patchKey": {
+			value: `
+value: {
+	apiVersion: "apps/v1"
+	kind:       "Deployment"
+	spec: template: spec: {
+		containers: [{
+			name: "test"
+		}]
+	}
+}
+patch: {
+	spec: template: spec: {
+		// +patchKey=name
+		containers: [{
+			name: "test"
+			env: [{
+				name:  "test-env"
+				value: "test-value"
+			}]
+		}]
+	}
+}
+`,
+			expectedErr: nil,
+			patchResult: `
+apiVersion: "apps/v1"
+kind:       "Deployment"
+spec: template: spec: {
+	containers: [{
+		name: "test"
+		env: [{
+			name:  "test-env"
+			value: "test-value"
+		}]
+	}]
+}
+`,
+		},
+		"test patch k8s object with patchStrategy": {
+			value: `
+value: {
+	apiVersion: "apps/v1"
+	kind:       "Deployment"
+	spec: template: metadata: {
+		name: "test-name"
+	}
+}
+patch: {
+	// +patchStrategy=retainKeys
+	spec: template: metadata: {
+		name: "test-patchStrategy"
+	}
+}
+`,
+			expectedErr: nil,
+			patchResult: `
+apiVersion: "apps/v1"
+kind:       "Deployment"
+spec: template: metadata: {
+	name: "test-patchStrategy"
+}
+`,
+		},
+	}
+
+	for name, tc := range testcases {
+		t.Run(name, func(t *testing.T) {
+			r := require.New(t)
+			v, err := value.NewValue(tc.value, nil, "")
+			r.NoError(err)
+			prd := &provider{}
+			err = prd.PatchK8sObject(nil, v, nil)
+			if tc.expectedErr != nil {
+				r.Equal(tc.expectedErr.Error(), err.Error())
+				return
+			}
+			r.NoError(err)
+			result, err := v.LookupValue("result")
+			r.NoError(err)
+			var patchResult map[string]interface{}
+			r.NoError(result.UnmarshalTo(&patchResult))
+			var expectResult map[string]interface{}
+			resultValue, err := value.NewValue(tc.patchResult, nil, "")
+			r.NoError(err)
+			r.NoError(resultValue.UnmarshalTo(&expectResult))
+			assert.Equal(t, expectResult, patchResult)
+		})
+	}
+}
+
+func TestConvertString(t *testing.T) {
+	testCases := map[string]struct {
+		from        string
+		expected    string
+		expectedErr error
+	}{
+		"success": {
+			from:     `bt: 'test'`,
+			expected: "test",
+		},
+		"fail": {
+			from:        `bt: 123`,
+			expectedErr: errors.New("bt: cannot use value 123 (type int) as string|bytes"),
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			r := require.New(t)
+			v, err := value.NewValue(tc.from, nil, "")
+			r.NoError(err)
+			prd := &provider{}
+			err = prd.String(nil, v, nil)
+			if tc.expectedErr != nil {
+				r.Equal(tc.expectedErr.Error(), err.Error())
+				return
+			}
+			r.NoError(err)
+			expected, err := v.LookupValue("str")
+			r.NoError(err)
+			ret, err := expected.CueValue().String()
+			r.NoError(err)
+			r.Equal(ret, tc.expected)
+		})
+	}
+}
+
+func TestInstall(t *testing.T) {
+	p := providers.NewProviders()
+	Install(p)
+	h, ok := p.GetHandler("util", "string")
+	r := require.New(t)
+	r.Equal(ok, true)
+	r.Equal(h != nil, true)
+}

pkg/workflow/tasks/discover.go

@@ -30,11 +30,11 @@ import (
 	"github.com/oam-dev/kubevela/pkg/velaql/providers/query"
 	wfContext "github.com/oam-dev/kubevela/pkg/workflow/context"
 	"github.com/oam-dev/kubevela/pkg/workflow/providers"
-	"github.com/oam-dev/kubevela/pkg/workflow/providers/convert"
 	"github.com/oam-dev/kubevela/pkg/workflow/providers/email"
 	"github.com/oam-dev/kubevela/pkg/workflow/providers/http"
 	"github.com/oam-dev/kubevela/pkg/workflow/providers/kube"
 	"github.com/oam-dev/kubevela/pkg/workflow/providers/time"
+	"github.com/oam-dev/kubevela/pkg/workflow/providers/util"
 	"github.com/oam-dev/kubevela/pkg/workflow/providers/workspace"
 	"github.com/oam-dev/kubevela/pkg/workflow/tasks/custom"
 	"github.com/oam-dev/kubevela/pkg/workflow/tasks/template"
@@ -77,8 +77,9 @@ func suspend(step v1beta1.WorkflowStep, opt *types.GeneratorOptions) (types.Task
 func NewTaskDiscover(providerHandlers providers.Providers, pd *packages.PackageDiscover, cli client.Client, dm discoverymapper.DiscoveryMapper) types.TaskDiscover {
 	// install builtin provider
 	workspace.Install(providerHandlers)
-	convert.Install(providerHandlers)
 	email.Install(providerHandlers)
+	util.Install(providerHandlers)
+
 	templateLoader := template.NewWorkflowStepTemplateLoader(cli, dm)
 	return &taskDiscover{
 		builtins: map[string]types.TaskGenerator{
@@ -123,7 +124,6 @@ func NewViewTaskDiscover(pd *packages.PackageDiscover, cli client.Client, cfg *r
 	time.Install(handlerProviders)
 	kube.Install(handlerProviders, cli, apply, delete)
 	http.Install(handlerProviders, cli, viewNs)
-	convert.Install(handlerProviders)
 	email.Install(handlerProviders)
 
 	templateLoader := template.NewViewTemplateLoader(cli, viewNs)