diff --git a/cmd/core/main.go b/cmd/core/main.go index 9c9929e17..551fcae88 100644 --- a/cmd/core/main.go +++ b/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 diff --git a/pkg/client/controller_client.go b/pkg/client/controller_client.go index 9bad4c5b7..867c4bcb7 100644 --- a/pkg/client/controller_client.go +++ b/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() }, }) 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 150842662..8db7f1efc 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 @@ -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 + ` ) diff --git a/pkg/monitor/metrics/application.go b/pkg/monitor/metrics/application.go index 10969acf0..cbab1b843 100644 --- a/pkg/monitor/metrics/application.go +++ b/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 -} diff --git a/pkg/resourcekeeper/componentrevision.go b/pkg/resourcekeeper/componentrevision.go index d82ba9440..cdc0acc27 100644 --- a/pkg/resourcekeeper/componentrevision.go +++ b/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 { diff --git a/pkg/resourcekeeper/dispatch.go b/pkg/resourcekeeper/dispatch.go index a23633740..a10f1f059 100644 --- a/pkg/resourcekeeper/dispatch.go +++ b/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) } diff --git a/pkg/resourcekeeper/gc_test.go b/pkg/resourcekeeper/gc_test.go index 0f9351598..01f676ad0 100644 --- a/pkg/resourcekeeper/gc_test.go +++ b/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) { diff --git a/pkg/resourcetracker/app.go b/pkg/resourcetracker/app.go index c9b84f1c8..dcaa5e9d4 100644 --- a/pkg/resourcetracker/app.go +++ b/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 diff --git a/pkg/resourcetracker/app_test.go b/pkg/resourcetracker/app_test.go index 466fba1d1..a07a2ee23 100644 --- a/pkg/resourcetracker/app_test.go +++ b/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++ { diff --git a/pkg/stdlib/op.cue b/pkg/stdlib/op.cue index b7300db5b..b6bcca589 100644 --- a/pkg/stdlib/op.cue +++ b/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" ... diff --git a/pkg/stdlib/pkgs/convert.cue b/pkg/stdlib/pkgs/convert.cue deleted file mode 100644 index 7b8f4a5bb..000000000 --- a/pkg/stdlib/pkgs/convert.cue +++ /dev/null @@ -1,8 +0,0 @@ -#String: { - #do: "string" - #provider: "convert" - - bt: bytes - str?: string - ... -} diff --git a/pkg/stdlib/pkgs/kube.cue b/pkg/stdlib/pkgs/kube.cue index 0e6727f09..4ca437636 100644 --- a/pkg/stdlib/pkgs/kube.cue +++ b/pkg/stdlib/pkgs/kube.cue @@ -6,6 +6,14 @@ ... } +#ApplyInParallel: { + #do: "apply-in-parallel" + #provider: "kube" + cluster: *"" | string + value: [...{...}] + ... +} + #Read: { #do: "read" #provider: "kube" diff --git a/pkg/stdlib/pkgs/util.cue b/pkg/stdlib/pkgs/util.cue new file mode 100644 index 000000000..e137d40ba --- /dev/null +++ b/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 + ... +} diff --git a/pkg/workflow/providers/convert/string.go b/pkg/workflow/providers/convert/string.go deleted file mode 100644 index e4715112e..000000000 --- a/pkg/workflow/providers/convert/string.go +++ /dev/null @@ -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, - }) -} diff --git a/pkg/workflow/providers/convert/string_test.go b/pkg/workflow/providers/convert/string_test.go deleted file mode 100644 index 1ed89c248..000000000 --- a/pkg/workflow/providers/convert/string_test.go +++ /dev/null @@ -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) -} diff --git a/pkg/workflow/providers/kube/handle.go b/pkg/workflow/providers/kube/handle.go index 7d6f15cbb..cd8f99390 100644 --- a/pkg/workflow/providers/kube/handle.go +++ b/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, }) } diff --git a/pkg/workflow/providers/kube/handle_test.go b/pkg/workflow/providers/kube/handle_test.go index a71fb3249..6cd77c9b6 100644 --- a/pkg/workflow/providers/kube/handle_test.go +++ b/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) { diff --git a/pkg/workflow/providers/oam/apply.go b/pkg/workflow/providers/oam/apply.go index fc9f10904..76276ff44 100644 --- a/pkg/workflow/providers/oam/apply.go +++ b/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" diff --git a/pkg/workflow/providers/util/util.go b/pkg/workflow/providers/util/util.go new file mode 100644 index 000000000..07cb768ed --- /dev/null +++ b/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, + }) +} diff --git a/pkg/workflow/providers/util/util_test.go b/pkg/workflow/providers/util/util_test.go new file mode 100644 index 000000000..d41a50edd --- /dev/null +++ b/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) +} diff --git a/pkg/workflow/tasks/discover.go b/pkg/workflow/tasks/discover.go index 8473a3634..702ba9048 100644 --- a/pkg/workflow/tasks/discover.go +++ b/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)