From 6cee4687437958cbaa34e41112c8a1734503ec3c Mon Sep 17 00:00:00 2001 From: yangsoon Date: Fri, 14 Jan 2022 15:18:02 +0800 Subject: [PATCH] Feat: add more performance optimization and prometheus metrics for controller (#3086) * Feat: add more prometheus metrics Signed-off-by: yangsoon * Feat: add detail gc rt duration metrics Signed-off-by: Yin Da * Feat: add monitor to client Signed-off-by: Yin Da * Feat: add all cache object Signed-off-by: Yin Da * Fix: watch job Signed-off-by: Yin Da * Feat: add deleg client Signed-off-by: Yin Da * Feat: add optimize for rt list and disable controllerrevision Signed-off-by: Yin Da * Feat: add apprev disable optimize Signed-off-by: Yin Da * Fix: optimize log Signed-off-by: Yin Da * Feat: add time recorder for app ctrl Signed-off-by: Yin Da * Feat: add in-memory workflow context Signed-off-by: Yin Da * Feat: add reconcile-reduction & random-pick-gc & optimize rt record Signed-off-by: Yin Da * Feat: add optimize for healthcheck & resourcetracker trigger Signed-off-by: Somefive * Chore: refactor Signed-off-by: Somefive * Feat: record the resource-tracker number by informer event-handler Signed-off-by: yangsoon * Feat: add promethus collect annotation in template Signed-off-by: yangsoon * Fix: command line comment bug Signed-off-by: Somefive * Chore: rename args and remove legacy controller metrics Signed-off-by: Somefive * Fix: make code reviewable Signed-off-by: yangsoon * Chore: optimize flag descriptions Signed-off-by: Somefive * Chore: break optimize package Signed-off-by: Somefive * Fix: gc policy test Signed-off-by: Somefive Co-authored-by: Yin Da Co-authored-by: yangsoon --- .../v1beta1/resourcetracker_types.go | 12 +- .../templates/kubevela-controller.yaml | 4 + .../templates/kubevela-controller.yaml | 4 + cmd/core/main.go | 6 +- pkg/appfile/appfile.go | 11 +- pkg/appfile/appfile_test.go | 2 +- pkg/appfile/parser.go | 8 + pkg/client/controller_client.go | 118 +++++++++++++++ pkg/client/delegating_client.go | 107 +++++++++++++ pkg/client/monitor_client.go | 142 ++++++++++++++++++ .../application/application_controller.go | 65 ++++++-- .../v1alpha2/application/apply.go | 9 ++ .../v1alpha2/application/gc_policy_test.go | 3 + .../v1alpha2/application/generator.go | 13 ++ .../v1alpha2/application/generator_test.go | 6 +- .../v1alpha2/application/revision.go | 48 +++++- pkg/controller/flags.go | 47 ++++++ pkg/monitor/context/context.go | 8 +- pkg/monitor/metrics/application.go | 132 ++++++++++++++++ pkg/monitor/metrics/workflow.go | 33 +++- pkg/multicluster/utils.go | 6 + pkg/resourcekeeper/gc.go | 33 +++- pkg/resourcekeeper/gc_test.go | 1 + pkg/resourcetracker/app.go | 11 +- pkg/resourcetracker/optimize.go | 69 +++++++++ pkg/workflow/context/context.go | 14 +- pkg/workflow/context/storage.go | 80 ++++++++++ pkg/workflow/workflow.go | 10 +- 28 files changed, 958 insertions(+), 44 deletions(-) create mode 100644 pkg/client/controller_client.go create mode 100644 pkg/client/delegating_client.go create mode 100644 pkg/client/monitor_client.go create mode 100644 pkg/controller/flags.go create mode 100644 pkg/monitor/metrics/application.go create mode 100644 pkg/resourcetracker/optimize.go create mode 100644 pkg/workflow/context/storage.go diff --git a/apis/core.oam.dev/v1beta1/resourcetracker_types.go b/apis/core.oam.dev/v1beta1/resourcetracker_types.go index 8ddf96e6d..6ff673e32 100644 --- a/apis/core.oam.dev/v1beta1/resourcetracker_types.go +++ b/apis/core.oam.dev/v1beta1/resourcetracker_types.go @@ -187,7 +187,7 @@ func (in *ResourceTracker) findMangedResourceIndex(mr ManagedResource) int { } // AddManagedResource add object to managed resources, if exists, update -func (in *ResourceTracker) AddManagedResource(rsc client.Object, metaOnly bool) { +func (in *ResourceTracker) AddManagedResource(rsc client.Object, metaOnly bool) (updated bool) { gvk := rsc.GetObjectKind().GroupVersionKind() mr := ManagedResource{ ClusterObjectReference: common.ClusterObjectReference{ @@ -206,17 +206,21 @@ func (in *ResourceTracker) AddManagedResource(rsc client.Object, metaOnly bool) mr.Data = &runtime.RawExtension{Object: rsc} } if idx := in.findMangedResourceIndex(mr); idx >= 0 { + if reflect.DeepEqual(in.Spec.ManagedResources[idx], mr) { + return false + } in.Spec.ManagedResources[idx] = mr } else { in.Spec.ManagedResources = append(in.Spec.ManagedResources, mr) } + return true } // DeleteManagedResource if remove flag is on, it will remove the object from recorded resources. // otherwise, it will mark the object as deleted instead of removing it // workflow stage: resources are marked as deleted (and execute the deletion action) // state-keep stage: resources marked as deleted and successfully deleted will be removed from resourcetracker -func (in *ResourceTracker) DeleteManagedResource(rsc client.Object, remove bool) { +func (in *ResourceTracker) DeleteManagedResource(rsc client.Object, remove bool) (updated bool) { gvk := rsc.GetObjectKind().GroupVersionKind() mr := ManagedResource{ ClusterObjectReference: common.ClusterObjectReference{ @@ -234,6 +238,9 @@ func (in *ResourceTracker) DeleteManagedResource(rsc client.Object, remove bool) if remove { in.Spec.ManagedResources = append(in.Spec.ManagedResources[:idx], in.Spec.ManagedResources[idx+1:]...) } else { + if reflect.DeepEqual(in.Spec.ManagedResources[idx], mr) { + return false + } in.Spec.ManagedResources[idx] = mr } } else { @@ -241,6 +248,7 @@ func (in *ResourceTracker) DeleteManagedResource(rsc client.Object, remove bool) in.Spec.ManagedResources = append(in.Spec.ManagedResources, mr) } } + return true } // addClusterObjectReference diff --git a/charts/vela-core/templates/kubevela-controller.yaml b/charts/vela-core/templates/kubevela-controller.yaml index 00b015a6f..31c8ff668 100644 --- a/charts/vela-core/templates/kubevela-controller.yaml +++ b/charts/vela-core/templates/kubevela-controller.yaml @@ -90,6 +90,10 @@ spec: metadata: labels: {{- include "kubevela.selectorLabels" . | nindent 8 }} + annotations: + prometheus.io/path: /metrics + prometheus.io/port: "8080" + prometheus.io/scrape: "true" spec: {{- with .Values.imagePullSecrets }} imagePullSecrets: diff --git a/charts/vela-minimal/templates/kubevela-controller.yaml b/charts/vela-minimal/templates/kubevela-controller.yaml index 9accfbaf6..8f85da4e7 100644 --- a/charts/vela-minimal/templates/kubevela-controller.yaml +++ b/charts/vela-minimal/templates/kubevela-controller.yaml @@ -92,6 +92,10 @@ spec: metadata: labels: {{- include "kubevela.selectorLabels" . | nindent 8 }} + annotations: + prometheus.io/path: /metrics + prometheus.io/port: "8080" + prometheus.io/scrape: "true" spec: {{- with .Values.imagePullSecrets }} imagePullSecrets: diff --git a/cmd/core/main.go b/cmd/core/main.go index 95e4ca9f1..3730b1950 100644 --- a/cmd/core/main.go +++ b/cmd/core/main.go @@ -30,13 +30,12 @@ import ( "strings" "time" - appsv1 "k8s.io/api/apps/v1" "k8s.io/klog/v2" "k8s.io/klog/v2/klogr" ctrl "sigs.k8s.io/controller-runtime" - "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/healthz" + ctrlClient "github.com/oam-dev/kubevela/pkg/client" standardcontroller "github.com/oam-dev/kubevela/pkg/controller" commonconfig "github.com/oam-dev/kubevela/pkg/controller/common" oamcontroller "github.com/oam-dev/kubevela/pkg/controller/core.oam.dev" @@ -130,6 +129,7 @@ func main() { flag.DurationVar(&retryPeriod, "leader-election-retry-period", 2*time.Second, "The duration the LeaderElector clients should wait between tries of actions") flag.BoolVar(&enableClusterGateway, "enable-cluster-gateway", false, "Enable cluster-gateway to use multicluster, disabled by default.") + standardcontroller.AddOptimizeFlags() flag.Parse() // setup logging @@ -213,7 +213,7 @@ func main() { // of controller-runtime. Additionally, set this value will affect not only application // controller but also all other controllers like definition controller. Therefore, for // functionalities like state-keep, they should be invented in other ways. - ClientDisableCacheFor: []client.Object{&appsv1.ControllerRevision{}}, + NewClient: ctrlClient.DefaultNewControllerClient, }) if err != nil { klog.ErrorS(err, "Unable to create a controller manager") diff --git a/pkg/appfile/appfile.go b/pkg/appfile/appfile.go index 5e7ad185b..5e9a485ed 100644 --- a/pkg/appfile/appfile.go +++ b/pkg/appfile/appfile.go @@ -45,6 +45,8 @@ import ( "github.com/oam-dev/kubevela/pkg/cue/model" "github.com/oam-dev/kubevela/pkg/cue/model/value" "github.com/oam-dev/kubevela/pkg/cue/process" + monitorContext "github.com/oam-dev/kubevela/pkg/monitor/context" + "github.com/oam-dev/kubevela/pkg/monitor/metrics" "github.com/oam-dev/kubevela/pkg/oam" "github.com/oam-dev/kubevela/pkg/oam/util" "github.com/oam-dev/kubevela/pkg/workflow/step" @@ -182,7 +184,14 @@ type Handler interface { } // PrepareWorkflowAndPolicy generates workflow steps and policies from an appFile -func (af *Appfile) PrepareWorkflowAndPolicy() ([]*unstructured.Unstructured, error) { +func (af *Appfile) PrepareWorkflowAndPolicy(ctx context.Context) ([]*unstructured.Unstructured, error) { + if ctx, ok := ctx.(monitorContext.Context); ok { + subCtx := ctx.Fork("prepare-workflow-and-policy", monitorContext.DurationMetric(func(v float64) { + metrics.PrepareWorkflowAndPolicyDurationHistogram.WithLabelValues("application").Observe(v) + })) + defer subCtx.Commit("finish prepare workflow and policy") + } + var externalPolicies []*unstructured.Unstructured var err error diff --git a/pkg/appfile/appfile_test.go b/pkg/appfile/appfile_test.go index 3aade2a77..1adbf6764 100644 --- a/pkg/appfile/appfile_test.go +++ b/pkg/appfile/appfile_test.go @@ -460,7 +460,7 @@ spec: } _, err := testAppfile.GenerateComponentManifests() Expect(err).Should(BeNil()) - gotPolicies, err := testAppfile.PrepareWorkflowAndPolicy() + gotPolicies, err := testAppfile.PrepareWorkflowAndPolicy(context.Background()) Expect(err).Should(BeNil()) Expect(len(gotPolicies)).ShouldNot(Equal(0)) diff --git a/pkg/appfile/parser.go b/pkg/appfile/parser.go index 4b5716926..1994bb6ca 100644 --- a/pkg/appfile/parser.go +++ b/pkg/appfile/parser.go @@ -32,6 +32,8 @@ import ( "github.com/oam-dev/kubevela/apis/types" "github.com/oam-dev/kubevela/pkg/cue/definition" "github.com/oam-dev/kubevela/pkg/cue/packages" + monitorContext "github.com/oam-dev/kubevela/pkg/monitor/context" + "github.com/oam-dev/kubevela/pkg/monitor/metrics" "github.com/oam-dev/kubevela/pkg/oam" "github.com/oam-dev/kubevela/pkg/oam/discoverymapper" "github.com/oam-dev/kubevela/pkg/oam/util" @@ -75,6 +77,12 @@ func NewDryRunApplicationParser(cli client.Client, dm discoverymapper.DiscoveryM // GenerateAppFile converts an application to an Appfile func (p *Parser) GenerateAppFile(ctx context.Context, app *v1beta1.Application) (*Appfile, error) { + if ctx, ok := ctx.(monitorContext.Context); ok { + subCtx := ctx.Fork("generate-app-file", monitorContext.DurationMetric(func(v float64) { + metrics.ParseAppFileDurationHistogram.WithLabelValues("application").Observe(v) + })) + defer subCtx.Commit("finish generate appFile") + } ns := app.Namespace appName := app.Name diff --git a/pkg/client/controller_client.go b/pkg/client/controller_client.go new file mode 100644 index 000000000..9bad4c5b7 --- /dev/null +++ b/pkg/client/controller_client.go @@ -0,0 +1,118 @@ +/* +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 client + +import ( + "context" + "strings" + "sync" + + "github.com/oam-dev/kubevela/pkg/oam" + "github.com/oam-dev/kubevela/pkg/resourcetracker" + + "github.com/pkg/errors" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/client-go/rest" + cache2 "k8s.io/client-go/tools/cache" + "sigs.k8s.io/controller-runtime/pkg/cache" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/apiutil" + + "github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1" + "github.com/oam-dev/kubevela/pkg/monitor/metrics" +) + +var ( + // CachedGVKs identifies the GVKs of resources to be cached during dispatching + CachedGVKs = "" + + rtCount = 0 + lock = sync.Mutex{} +) + +// DefaultNewControllerClient function for creating controller client +func DefaultNewControllerClient(cache cache.Cache, config *rest.Config, options client.Options, uncachedObjects ...client.Object) (c client.Client, err error) { + rawClient, err := client.New(config, options) + if err != nil { + return nil, errors.Wrapf(err, "failed to get raw client") + } + + mClient := &monitorClient{rawClient} + if err := resourcetracker.AddResourceTrackerCacheIndex(cache); err != nil { + return nil, errors.Wrapf(err, "failed to add app index to ResourceTracker cache") + } + mCache := &monitorCache{cache} + + uncachedStructuredGVKs := map[schema.GroupVersionKind]struct{}{} + for _, obj := range uncachedObjects { + gvk, err := apiutil.GVKForObject(obj, mClient.Scheme()) + if err != nil { + return nil, err + } + uncachedStructuredGVKs[gvk] = struct{}{} + } + + cachedUnstructuredGVKs := map[schema.GroupVersionKind]struct{}{} + for _, s := range strings.Split(CachedGVKs, ",") { + s = strings.Trim(s, " ") + if len(s) > 0 { + gvk, _ := schema.ParseKindArg(s) + if gvk == nil { + return nil, errors.Errorf("invalid cached gvk: %s", s) + } + cachedUnstructuredGVKs[*gvk] = struct{}{} + } + } + + informer, err := cache.GetInformerForKind(context.Background(), v1beta1.ResourceTrackerKindVersionKind) + if err != nil { + return nil, err + } + + informer.AddEventHandler(cache2.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + lock.Lock() + rtCount++ + metrics.ResourceTrackerNumberGauge.WithLabelValues( + metrics.ExtractMetricValuesFromObjectLabel(obj, oam.LabelAppName, oam.LabelAppNamespace)...).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)) + lock.Unlock() + }, + }) + + dClient := &delegatingClient{ + scheme: mClient.Scheme(), + mapper: mClient.RESTMapper(), + Reader: &delegatingReader{ + CacheReader: mCache, + ClientReader: mClient, + scheme: mClient.Scheme(), + uncachedStructuredGVKs: uncachedStructuredGVKs, + cachedUnstructuredGVKs: cachedUnstructuredGVKs, + }, + Writer: mClient, + StatusClient: mClient, + } + + return dClient, nil +} diff --git a/pkg/client/delegating_client.go b/pkg/client/delegating_client.go new file mode 100644 index 000000000..f58ffbf42 --- /dev/null +++ b/pkg/client/delegating_client.go @@ -0,0 +1,107 @@ +/* +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 client + +import ( + "context" + "strings" + + "k8s.io/apimachinery/pkg/api/meta" + "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/client/apiutil" + + "github.com/oam-dev/kubevela/pkg/multicluster" + "github.com/oam-dev/kubevela/pkg/resourcetracker" +) + +type delegatingClient struct { + client.Reader + client.Writer + client.StatusClient + + scheme *runtime.Scheme + mapper meta.RESTMapper +} + +// Scheme returns the scheme this client is using. +func (d *delegatingClient) Scheme() *runtime.Scheme { + return d.scheme +} + +// RESTMapper returns the rest mapper this client is using. +func (d *delegatingClient) RESTMapper() meta.RESTMapper { + return d.mapper +} + +// delegatingReader extend the delegatingReader from controller-runtime/pkg/client +// 1. for requests not in local cluster, disable cache +// 2. for structured types, inherit the cache blacklist +// 3. for unstructured types, use cache whitelist +type delegatingReader struct { + CacheReader client.Reader + ClientReader client.Reader + + uncachedStructuredGVKs map[schema.GroupVersionKind]struct{} + cachedUnstructuredGVKs map[schema.GroupVersionKind]struct{} + scheme *runtime.Scheme +} + +func (d *delegatingReader) shouldBypassCache(ctx context.Context, obj runtime.Object) (bool, error) { + // non-local resource cannot use cache + if !multicluster.IsInLocalCluster(ctx) { + return true, nil + } + gvk, err := apiutil.GVKForObject(obj, d.scheme) + if err != nil { + return false, err + } + if meta.IsListType(obj) { + gvk.Kind = strings.TrimSuffix(gvk.Kind, "List") + } + _, isUnstructured := obj.(*unstructured.Unstructured) + _, isUnstructuredList := obj.(*unstructured.UnstructuredList) + if isUnstructured || isUnstructuredList { + _, shouldCache := d.cachedUnstructuredGVKs[gvk] + return !shouldCache, nil + } + _, shouldNotCache := d.uncachedStructuredGVKs[gvk] + return shouldNotCache, nil +} + +// Get retrieves an obj for a given object key from the Kubernetes Cluster. +func (d *delegatingReader) Get(ctx context.Context, key client.ObjectKey, obj client.Object) error { + if isUncached, err := d.shouldBypassCache(ctx, obj); err != nil { + return err + } else if isUncached { + return d.ClientReader.Get(ctx, key, obj) + } + return d.CacheReader.Get(ctx, key, obj) +} + +// List retrieves list of objects for a given namespace and list options. +func (d *delegatingReader) List(ctx context.Context, list client.ObjectList, opts ...client.ListOption) error { + opts = resourcetracker.ExtendResourceTrackerListOption(list, opts) + if isUncached, err := d.shouldBypassCache(ctx, list); err != nil { + return err + } else if isUncached { + return d.ClientReader.List(ctx, list, opts...) + } + return d.CacheReader.List(ctx, list, opts...) +} diff --git a/pkg/client/monitor_client.go b/pkg/client/monitor_client.go new file mode 100644 index 000000000..60429fa41 --- /dev/null +++ b/pkg/client/monitor_client.go @@ -0,0 +1,142 @@ +/* +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 client + +import ( + "context" + "reflect" + "strings" + "time" + + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/cache" + "sigs.k8s.io/controller-runtime/pkg/client" + + "github.com/oam-dev/kubevela/pkg/monitor/metrics" + "github.com/oam-dev/kubevela/pkg/multicluster" +) + +func monitor(ctx context.Context, verb string, obj runtime.Object) func() { + o := obj.GetObjectKind().GroupVersionKind() + _, isUnstructured := obj.(*unstructured.Unstructured) + _, isUnstructuredList := obj.(*unstructured.UnstructuredList) + un := "structured" + if isUnstructured || isUnstructuredList { + un = "unstructured" + } + clusterName := multicluster.ClusterNameInContext(ctx) + if clusterName == "" { + clusterName = multicluster.ClusterLocalName + } + kind := o.Kind + if kind == "" { + if t := reflect.TypeOf(obj); t.Kind() == reflect.Ptr { + kind = t.Elem().Name() + } else { + kind = t.Name() + } + } + kind = strings.TrimSuffix(kind, "List") + begin := time.Now() + return func() { + v := time.Since(begin).Seconds() + metrics.ClientRequestHistogram.WithLabelValues(verb, kind, o.GroupVersion().String(), un, clusterName).Observe(v) + } +} + +type monitorCache struct { + cache.Cache +} + +func (c *monitorCache) Get(ctx context.Context, key client.ObjectKey, obj client.Object) error { + cb := monitor(ctx, "GetCache", obj) + defer cb() + return c.Cache.Get(ctx, key, obj) +} + +func (c *monitorCache) List(ctx context.Context, list client.ObjectList, opts ...client.ListOption) error { + cb := monitor(ctx, "ListCache", list) + defer cb() + return c.Cache.List(ctx, list, opts...) +} + +type monitorClient struct { + client.Client +} + +func (c *monitorClient) Get(ctx context.Context, key client.ObjectKey, obj client.Object) error { + cb := monitor(ctx, "Get", obj) + defer cb() + return c.Client.Get(ctx, key, obj) +} + +func (c *monitorClient) List(ctx context.Context, list client.ObjectList, opts ...client.ListOption) error { + cb := monitor(ctx, "List", list) + defer cb() + return c.Client.List(ctx, list, opts...) +} + +func (c *monitorClient) Create(ctx context.Context, obj client.Object, opts ...client.CreateOption) error { + cb := monitor(ctx, "Create", obj) + defer cb() + return c.Client.Create(ctx, obj, opts...) +} + +func (c *monitorClient) Delete(ctx context.Context, obj client.Object, opts ...client.DeleteOption) error { + cb := monitor(ctx, "Delete", obj) + defer cb() + return c.Client.Delete(ctx, obj, opts...) +} + +func (c *monitorClient) Update(ctx context.Context, obj client.Object, opts ...client.UpdateOption) error { + cb := monitor(ctx, "Update", obj) + defer cb() + return c.Client.Update(ctx, obj, opts...) +} + +func (c *monitorClient) Patch(ctx context.Context, obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + cb := monitor(ctx, "Patch", obj) + defer cb() + return c.Client.Patch(ctx, obj, patch, opts...) +} + +func (c *monitorClient) DeleteAllOf(ctx context.Context, obj client.Object, opts ...client.DeleteAllOfOption) error { + cb := monitor(ctx, "DeleteAllOf", obj) + defer cb() + return c.Client.DeleteAllOf(ctx, obj, opts...) +} + +func (c *monitorClient) Status() client.StatusWriter { + return &monitorStatusWriter{c.Client.Status()} +} + +type monitorStatusWriter struct { + client.StatusWriter +} + +func (w *monitorStatusWriter) Update(ctx context.Context, obj client.Object, opts ...client.UpdateOption) error { + cb := monitor(ctx, "StatusUpdate", obj) + defer cb() + return w.StatusWriter.Update(ctx, obj, opts...) +} + +func (w *monitorStatusWriter) Patch(ctx context.Context, obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + cb := monitor(ctx, "StatusPatch", obj) + defer cb() + return w.StatusWriter.Patch(ctx, obj, patch, opts...) +} 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 ee444a8a9..4795cf1fa 100644 --- a/pkg/controller/core.oam.dev/v1alpha2/application/application_controller.go +++ b/pkg/controller/core.oam.dev/v1alpha2/application/application_controller.go @@ -22,6 +22,9 @@ import ( "reflect" "time" + "github.com/crossplane/crossplane-runtime/pkg/event" + "github.com/crossplane/crossplane-runtime/pkg/meta" + "github.com/pkg/errors" corev1 "k8s.io/api/core/v1" kerrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -36,10 +39,6 @@ import ( "sigs.k8s.io/controller-runtime/pkg/reconcile" "sigs.k8s.io/controller-runtime/pkg/source" - "github.com/crossplane/crossplane-runtime/pkg/event" - "github.com/crossplane/crossplane-runtime/pkg/meta" - "github.com/pkg/errors" - "github.com/oam-dev/kubevela/apis/core.oam.dev/common" "github.com/oam-dev/kubevela/apis/core.oam.dev/condition" "github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1" @@ -49,12 +48,14 @@ import ( core "github.com/oam-dev/kubevela/pkg/controller/core.oam.dev" "github.com/oam-dev/kubevela/pkg/cue/packages" monitorContext "github.com/oam-dev/kubevela/pkg/monitor/context" + "github.com/oam-dev/kubevela/pkg/monitor/metrics" "github.com/oam-dev/kubevela/pkg/oam" "github.com/oam-dev/kubevela/pkg/oam/discoverymapper" oamutil "github.com/oam-dev/kubevela/pkg/oam/util" "github.com/oam-dev/kubevela/pkg/resourcekeeper" "github.com/oam-dev/kubevela/pkg/resourcetracker" "github.com/oam-dev/kubevela/pkg/workflow" + wfContext "github.com/oam-dev/kubevela/pkg/workflow/context" "github.com/oam-dev/kubevela/version" ) @@ -70,6 +71,13 @@ const ( resourceTrackerFinalizer = "app.oam.dev/resource-tracker-finalizer" ) +var ( + // EnableReconcileLoopReduction optimize application reconcile loop by fusing phase transition + EnableReconcileLoopReduction = false + // EnableResourceTrackerDeleteOnlyTrigger optimize ResourceTracker mutate event trigger by only receiving deleting events + EnableResourceTrackerDeleteOnlyTrigger = true +) + // Reconciler reconciles an Application object type Reconciler struct { client.Client @@ -91,8 +99,8 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu defer cancel() logCtx := monitorContext.NewTraceContext(ctx, "").AddTag("application", req.String(), "controller", "application") - logCtx.Info("Reconcile application") - defer logCtx.Commit("Reconcile application") + logCtx.Info("Start reconcile application") + defer logCtx.Commit("End reconcile application") app := new(v1beta1.Application) if err := r.Get(ctx, client.ObjectKey{ Name: req.Name, @@ -104,7 +112,10 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu return r.result(client.IgnoreNotFound(err)).ret() } - logCtx.AddTag("resource_version", app.ResourceVersion) + timeReporter := timeReconcile(app) + defer timeReporter() + + logCtx.AddTag("resource_version", app.ResourceVersion).AddTag("generation", app.Generation) ctx = oamutil.SetNamespaceInCtx(ctx, app.Namespace) logCtx.SetContext(ctx) if annotations := app.GetAnnotations(); annotations == nil || annotations[oam.AnnotationKubeVelaVersion] == "" { @@ -128,7 +139,6 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu appFile, err := appParser.GenerateAppFile(logCtx, app) if err != nil { - logCtx.Error(err, "Failed to parse application") r.Recorder.Event(app, event.Warning(velatypes.ReasonFailedParse, err)) return r.endWithNegativeCondition(logCtx, app, condition.ErrorCondition("Parsed", err), common.ApplicationRendering) } @@ -156,7 +166,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu } logCtx.Info("Successfully apply application revision") - externalPolicies, err := appFile.PrepareWorkflowAndPolicy() + externalPolicies, err := appFile.PrepareWorkflowAndPolicy(logCtx) if err != nil { logCtx.Error(err, "[Handle PrepareWorkflowAndPolicy]") r.Recorder.Event(app, event.Warning(velatypes.ReasonFailedRender, err)) @@ -220,7 +230,9 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu app.Status.SetConditions(condition.ReadyCondition(common.WorkflowCondition.String())) r.Recorder.Event(app, event.Normal(velatypes.ReasonApplied, velatypes.MessageWorkflowFinished)) logCtx.Info("Application manifests has applied by workflow successfully") - return r.gcResourceTrackers(logCtx, handler, common.ApplicationWorkflowFinished, false) + if !EnableReconcileLoopReduction { + return r.gcResourceTrackers(logCtx, handler, common.ApplicationWorkflowFinished, false) + } case common.WorkflowStateFinished: logCtx.Info("Workflow state=Finished") if status := app.Status.Workflow; status != nil && status.Terminated { @@ -258,6 +270,11 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu } func (r *Reconciler) gcResourceTrackers(logCtx monitorContext.Context, handler *AppHandler, phase common.ApplicationPhase, gcOutdated bool) (ctrl.Result, error) { + subCtx := logCtx.Fork("gc_resourceTrackers", monitorContext.DurationMetric(func(v float64) { + metrics.GCResourceTrackersDurationHistogram.WithLabelValues("-").Observe(v) + })) + defer subCtx.Commit("finish gc resourceTrackers") + var options []resourcekeeper.GCOption if !gcOutdated { options = append(options, resourcekeeper.DisableMarkStageGCOption{}, resourcekeeper.DisableGCComponentRevisionOption{}, resourcekeeper.DisableLegacyGCOption{}) @@ -318,12 +335,21 @@ func (r *Reconciler) result(err error) *reconcileResult { func (r *Reconciler) handleFinalizers(ctx monitorContext.Context, app *v1beta1.Application, handler *AppHandler) (bool, ctrl.Result, error) { if app.ObjectMeta.DeletionTimestamp.IsZero() { if !meta.FinalizerExists(app, resourceTrackerFinalizer) { + subCtx := ctx.Fork("handle-finalizers", monitorContext.DurationMetric(func(v float64) { + metrics.HandleFinalizersDurationHistogram.WithLabelValues("application", "add").Observe(v) + })) + defer subCtx.Commit("finish add finalizers") meta.AddFinalizer(app, resourceTrackerFinalizer) - ctx.Info("Register new finalizer for application", "finalizer", resourceTrackerFinalizer) - return r.result(errors.Wrap(r.Client.Update(ctx, app), errUpdateApplicationFinalizer)).end(true) + subCtx.Info("Register new finalizer for application", "finalizer", resourceTrackerFinalizer) + endReconcile := !EnableReconcileLoopReduction + return r.result(errors.Wrap(r.Client.Update(ctx, app), errUpdateApplicationFinalizer)).end(endReconcile) } } else { if meta.FinalizerExists(app, resourceTrackerFinalizer) { + subCtx := ctx.Fork("handle-finalizers", monitorContext.DurationMetric(func(v float64) { + metrics.HandleFinalizersDurationHistogram.WithLabelValues("application", "remove").Observe(v) + })) + defer subCtx.Commit("finish remove finalizers") rootRT, currentRT, historyRTs, cvRT, err := resourcetracker.ListApplicationResourceTrackers(ctx, r.Client, app) if err != nil { return r.result(err).end(true) @@ -336,6 +362,9 @@ func (r *Reconciler) handleFinalizers(ctx monitorContext.Context, app *v1beta1.A meta.RemoveFinalizer(app, resourceTrackerFinalizer) return r.result(errors.Wrap(r.Client.Update(ctx, app), errUpdateApplicationFinalizer)).end(true) } + if wfContext.EnableInMemoryContext { + wfContext.MemStore.DeleteInMemoryContext(app.Name) + } return true, result, err } } @@ -506,6 +535,9 @@ func filterManagedFieldChangesUpdate(e ctrlEvent.UpdateEvent) bool { func handleResourceTracker(obj client.Object, limitingInterface workqueue.RateLimitingInterface) { rt, ok := obj.(*v1beta1.ResourceTracker) if ok { + if EnableResourceTrackerDeleteOnlyTrigger && rt.GetDeletionTimestamp() == nil { + return + } if labels := rt.Labels; labels != nil { var request reconcile.Request request.Name = labels[oam.LabelAppName] @@ -516,3 +548,12 @@ func handleResourceTracker(obj client.Object, limitingInterface workqueue.RateLi } } } + +func timeReconcile(app *v1beta1.Application) func() { + t := time.Now() + beginPhase := string(app.Status.Phase) + return func() { + v := time.Since(t).Seconds() + metrics.ApplicationReconcileTimeHistogram.WithLabelValues(beginPhase, string(app.Status.Phase)).Observe(v) + } +} diff --git a/pkg/controller/core.oam.dev/v1alpha2/application/apply.go b/pkg/controller/core.oam.dev/v1alpha2/application/apply.go index c04a4a09e..7fadfdf37 100644 --- a/pkg/controller/core.oam.dev/v1alpha2/application/apply.go +++ b/pkg/controller/core.oam.dev/v1alpha2/application/apply.go @@ -31,6 +31,9 @@ import ( "github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1" "github.com/oam-dev/kubevela/apis/types" "github.com/oam-dev/kubevela/pkg/appfile" + + monitorContext "github.com/oam-dev/kubevela/pkg/monitor/context" + "github.com/oam-dev/kubevela/pkg/monitor/metrics" "github.com/oam-dev/kubevela/pkg/multicluster" "github.com/oam-dev/kubevela/pkg/resourcekeeper" ) @@ -54,6 +57,12 @@ type AppHandler struct { // NewAppHandler create new app handler func NewAppHandler(ctx context.Context, r *Reconciler, app *v1beta1.Application, parser *appfile.Parser) (*AppHandler, error) { + if ctx, ok := ctx.(monitorContext.Context); ok { + subCtx := ctx.Fork("create-app-handler", monitorContext.DurationMetric(func(v float64) { + metrics.CreateAppHandlerDurationHistogram.WithLabelValues("application").Observe(v) + })) + defer subCtx.Commit("finish create appHandler") + } resourceHandler, err := resourcekeeper.NewResourceKeeper(ctx, r.Client, app) if err != nil { return nil, errors.Wrapf(err, "failed to create resourceKeeper") diff --git a/pkg/controller/core.oam.dev/v1alpha2/application/gc_policy_test.go b/pkg/controller/core.oam.dev/v1alpha2/application/gc_policy_test.go index c085bf226..7a4494109 100644 --- a/pkg/controller/core.oam.dev/v1alpha2/application/gc_policy_test.go +++ b/pkg/controller/core.oam.dev/v1alpha2/application/gc_policy_test.go @@ -44,6 +44,7 @@ import ( "github.com/oam-dev/kubevela/pkg/oam" "github.com/oam-dev/kubevela/pkg/oam/testutil" "github.com/oam-dev/kubevela/pkg/oam/util" + "github.com/oam-dev/kubevela/pkg/resourcekeeper" ) var _ = Describe("Test Application with GC options", func() { @@ -107,6 +108,7 @@ var _ = Describe("Test Application with GC options", func() { } It("Each update will create a new workload and trait object", func() { + resourcekeeper.MarkWithProbability = 1.0 app := baseApp.DeepCopy() app.SetNamespace(ns.Name) app.SetName("app-with-worker-ingress") @@ -391,6 +393,7 @@ var _ = Describe("Test Application with GC options", func() { }) It("Each update will only update workload", func() { + resourcekeeper.MarkWithProbability = 1.0 app := baseApp.DeepCopy() app.Spec.Components[0].Traits = nil app.Spec.Components[0].Name = "only-work" diff --git a/pkg/controller/core.oam.dev/v1alpha2/application/generator.go b/pkg/controller/core.oam.dev/v1alpha2/application/generator.go index be3d03cad..c0c70a4bf 100644 --- a/pkg/controller/core.oam.dev/v1alpha2/application/generator.go +++ b/pkg/controller/core.oam.dev/v1alpha2/application/generator.go @@ -19,6 +19,7 @@ import ( "context" "encoding/json" "strings" + "time" "github.com/crossplane/crossplane-runtime/pkg/meta" "github.com/pkg/errors" @@ -31,6 +32,7 @@ import ( "github.com/oam-dev/kubevela/pkg/appfile" "github.com/oam-dev/kubevela/pkg/controller/core.oam.dev/v1alpha2/application/assemble" "github.com/oam-dev/kubevela/pkg/cue/model/value" + "github.com/oam-dev/kubevela/pkg/monitor/metrics" "github.com/oam-dev/kubevela/pkg/multicluster" "github.com/oam-dev/kubevela/pkg/oam" "github.com/oam-dev/kubevela/pkg/oam/util" @@ -46,6 +48,11 @@ import ( wfTypes "github.com/oam-dev/kubevela/pkg/workflow/types" ) +var ( + // DisableResourceApplyDoubleCheck optimize applyComponentFunc by disable post resource existing check after dispatch + DisableResourceApplyDoubleCheck = false +) + // GenerateApplicationSteps generate application steps. // nolint:gocyclo func (h *AppHandler) GenerateApplicationSteps(ctx context.Context, @@ -143,6 +150,9 @@ func (h *AppHandler) renderComponentFunc(appParser *appfile.Parser, appRev *v1be func (h *AppHandler) applyComponentFunc(appParser *appfile.Parser, appRev *v1beta1.ApplicationRevision, af *appfile.Appfile) oamProvider.ComponentApply { return func(comp common.ApplicationComponent, patcher *value.Value, clusterName string, overrideNamespace string, env string) (*unstructured.Unstructured, []*unstructured.Unstructured, bool, error) { + t := time.Now() + defer func() { metrics.ApplyComponentTimeHistogram.WithLabelValues("-").Observe(time.Since(t).Seconds()) }() + ctx := multicluster.ContextWithClusterName(context.Background(), clusterName) ctx = contextWithComponentRevisionNamespace(ctx, overrideNamespace) ctx = envbinding.ContextWithEnvName(ctx, env) @@ -181,6 +191,9 @@ func (h *AppHandler) applyComponentFunc(appParser *appfile.Parser, appRev *v1bet if !isHealth { return nil, nil, false, nil } + if DisableResourceApplyDoubleCheck { + return readyWorkload, readyTraits, true, nil + } workload, traits, err := getComponentResources(ctx, manifest, wl.SkipApplyWorkload, h.r.Client) return workload, traits, true, err } diff --git a/pkg/controller/core.oam.dev/v1alpha2/application/generator_test.go b/pkg/controller/core.oam.dev/v1alpha2/application/generator_test.go index 6396f495e..8f19ce030 100644 --- a/pkg/controller/core.oam.dev/v1alpha2/application/generator_test.go +++ b/pkg/controller/core.oam.dev/v1alpha2/application/generator_test.go @@ -107,7 +107,7 @@ var _ = Describe("Test Application workflow generator", func() { } af, err := appParser.GenerateAppFile(ctx, app) Expect(err).Should(BeNil()) - _, err = af.PrepareWorkflowAndPolicy() + _, err = af.PrepareWorkflowAndPolicy(context.Background()) Expect(err).Should(BeNil()) appRev := &oamcore.ApplicationRevision{} @@ -148,7 +148,7 @@ var _ = Describe("Test Application workflow generator", func() { } af, err := appParser.GenerateAppFile(ctx, app) Expect(err).Should(BeNil()) - _, err = af.PrepareWorkflowAndPolicy() + _, err = af.PrepareWorkflowAndPolicy(context.Background()) Expect(err).Should(BeNil()) appRev := &oamcore.ApplicationRevision{} @@ -204,7 +204,7 @@ var _ = Describe("Test Application workflow generator", func() { } af, err := appParser.GenerateAppFile(ctx, app) Expect(err).Should(BeNil()) - _, err = af.PrepareWorkflowAndPolicy() + _, err = af.PrepareWorkflowAndPolicy(context.Background()) Expect(err).Should(BeNil()) apprev := &oamcore.ApplicationRevision{ ObjectMeta: metav1.ObjectMeta{ diff --git a/pkg/controller/core.oam.dev/v1alpha2/application/revision.go b/pkg/controller/core.oam.dev/v1alpha2/application/revision.go index f6e09bd7d..f565ec0e6 100644 --- a/pkg/controller/core.oam.dev/v1alpha2/application/revision.go +++ b/pkg/controller/core.oam.dev/v1alpha2/application/revision.go @@ -36,10 +36,6 @@ import ( "k8s.io/utils/pointer" "sigs.k8s.io/controller-runtime/pkg/client" - "github.com/oam-dev/kubevela/pkg/cue/model" - "github.com/oam-dev/kubevela/pkg/multicluster" - "github.com/oam-dev/kubevela/pkg/policy/envbinding" - "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" @@ -47,8 +43,13 @@ import ( "github.com/oam-dev/kubevela/pkg/appfile" helmapi "github.com/oam-dev/kubevela/pkg/appfile/helm/flux2apis" "github.com/oam-dev/kubevela/pkg/controller/utils" + "github.com/oam-dev/kubevela/pkg/cue/model" + monitorContext "github.com/oam-dev/kubevela/pkg/monitor/context" + "github.com/oam-dev/kubevela/pkg/monitor/metrics" + "github.com/oam-dev/kubevela/pkg/multicluster" "github.com/oam-dev/kubevela/pkg/oam" "github.com/oam-dev/kubevela/pkg/oam/util" + "github.com/oam-dev/kubevela/pkg/policy/envbinding" ) type contextKey string @@ -68,6 +69,13 @@ const ( ComponentRevisionNamespaceContextKey = contextKey("component-revision-namespace") ) +var ( + // DisableAllComponentRevision disable component revision creation + DisableAllComponentRevision = false + // DisableAllApplicationRevision disable application revision creation + DisableAllApplicationRevision = false +) + func contextWithComponentRevisionNamespace(ctx context.Context, ns string) context.Context { return context.WithValue(ctx, ComponentRevisionNamespaceContextKey, ns) } @@ -141,6 +149,12 @@ func SprintComponentManifest(cm *types.ComponentManifest) string { // PrepareCurrentAppRevision will generate a pure revision without metadata and rendered result // the generated revision will be compare with the last revision to see if there's any difference. func (h *AppHandler) PrepareCurrentAppRevision(ctx context.Context, af *appfile.Appfile) error { + if ctx, ok := ctx.(monitorContext.Context); ok { + subCtx := ctx.Fork("prepare-current-appRevision", monitorContext.DurationMetric(func(v float64) { + metrics.PrepareCurrentAppRevisionDurationHistogram.WithLabelValues("application").Observe(v) + })) + defer subCtx.Commit("finish prepare current appRevision") + } appRev, appRevisionHash, err := h.gatherRevisionSpec(af) if err != nil { return err @@ -243,6 +257,9 @@ func (h *AppHandler) gatherRevisionSpec(af *appfile.Appfile) (*v1beta1.Applicati } func (h *AppHandler) getLatestAppRevision(ctx context.Context) error { + if DisableAllApplicationRevision { + return nil + } if h.app.Status.LatestRevision == nil || len(h.app.Status.LatestRevision.Name) == 0 { return nil } @@ -429,6 +446,10 @@ func DeepEqualRevision(old, new *v1beta1.ApplicationRevision) bool { // 1. if update component create a new component Revision // 2. check all componentTrait rely on componentRevName, if yes fill it func (h *AppHandler) HandleComponentsRevision(ctx context.Context, compManifests []*types.ComponentManifest) error { + if DisableAllComponentRevision { + return nil + } + for _, cm := range compManifests { // external revision specified @@ -653,6 +674,16 @@ func componentManifest2Component(cm *types.ComponentManifest) (*v1alpha2.Compone // FinalizeAndApplyAppRevision finalise AppRevision object and apply it func (h *AppHandler) FinalizeAndApplyAppRevision(ctx context.Context) error { + if DisableAllApplicationRevision { + return nil + } + + if ctx, ok := ctx.(monitorContext.Context); ok { + subCtx := ctx.Fork("apply-app-revision", monitorContext.DurationMetric(func(v float64) { + metrics.ApplyAppRevisionDurationHistogram.WithLabelValues("application").Observe(v) + })) + defer subCtx.Commit("finish apply app revision") + } appRev := h.currentAppRev appRev.Namespace = h.app.Namespace appRev.SetGroupVersionKind(v1beta1.ApplicationRevisionGroupVersionKind) @@ -688,6 +719,9 @@ func (h *AppHandler) FinalizeAndApplyAppRevision(ctx context.Context) error { // UpdateAppLatestRevisionStatus only call to update app's latest revision status after applying manifests successfully // otherwise it will override previous revision which is used during applying to do GC jobs func (h *AppHandler) UpdateAppLatestRevisionStatus(ctx context.Context) error { + if DisableAllApplicationRevision { + return nil + } if !h.isNewRevision { // skip update if app revision is not changed return nil @@ -711,6 +745,9 @@ func (h *AppHandler) UpdateAppLatestRevisionStatus(ctx context.Context) error { // cleanUpApplicationRevision check all appRevisions of the application, remove them if the number of them exceed the limit func cleanUpApplicationRevision(ctx context.Context, h *AppHandler) error { + if DisableAllApplicationRevision { + return nil + } listOpts := []client.ListOption{ client.InNamespace(h.app.Namespace), client.MatchingLabels{oam.LabelAppName: h.app.Name}, @@ -795,6 +832,9 @@ func (h historiesByRevision) Less(i, j int) bool { } func cleanUpWorkflowComponentRevision(ctx context.Context, h *AppHandler) error { + if DisableAllComponentRevision { + return nil + } // collect component revision in use compRevisionInUse := map[string]map[string]struct{}{} for _, resource := range h.app.Status.AppliedResources { diff --git a/pkg/controller/flags.go b/pkg/controller/flags.go new file mode 100644 index 000000000..ecd82380d --- /dev/null +++ b/pkg/controller/flags.go @@ -0,0 +1,47 @@ +/* +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 controller + +import ( + "flag" + + ctrlClient "github.com/oam-dev/kubevela/pkg/client" + "github.com/oam-dev/kubevela/pkg/controller/core.oam.dev/v1alpha2/application" + "github.com/oam-dev/kubevela/pkg/resourcekeeper" + "github.com/oam-dev/kubevela/pkg/resourcetracker" + "github.com/oam-dev/kubevela/pkg/workflow" + wfContext "github.com/oam-dev/kubevela/pkg/workflow/context" +) + +// AddOptimizeFlags add flags +func AddOptimizeFlags() { + // optimize client + flag.StringVar(&ctrlClient.CachedGVKs, "optimize-cached-gvks", "", "Types of resources to be cached. For example, --optimize-cached-gvks=Deployment.v1.apps,Job.v1.batch . If you have dedicated resources to be managed in your system, you can turn it on to improve performance. NOTE: this optimization only works for single-cluster.") + flag.BoolVar(&resourcetracker.OptimizeListOp, "optimize-resource-tracker-list-op", true, "Optimize ResourceTracker List Op by adding index. This will increase the use of memory and accelerate the list operation of ResourceTracker. Default to enable it . If you want to reduce the memory use of KubeVela, you can switch it off.") + + // optimize controller reconcile loop + flag.BoolVar(&application.EnableReconcileLoopReduction, "optimize-controller-reconcile-loop-reduction", false, "Optimize ApplicationController reconcile by reducing the number of loops to reconcile application. In detail, reconciles after finalizer patching and workflow finished will not return immediately but will continue running. If you do not care about the occasional re-run of workflow, you can switch it on to further improve KubeVela controller performance.") + + // optimize functions + flag.Float64Var(&resourcekeeper.MarkWithProbability, "optimize-mark-with-prob", 0.1, "Optimize ResourceTracker GC by only run mark with probability. Side effect: outdated ResourceTracker might not be able to be removed immediately. Default to 0.1. If you want to cleanup outdated resource for keepLegacyResource mode immediately, set it to 1.0 to disable this optimization.") + flag.BoolVar(&application.DisableAllComponentRevision, "optimize-disable-component-revision", false, "Optimize ComponentRevision by disabling the creation and gc. Side effect: rollout cannot be used. If you don't use rollout trait, you can switch it on to reduce the storage and improve performance.") + flag.BoolVar(&application.DisableAllApplicationRevision, "optimize-disable-application-revision", false, "Optimize ApplicationRevision by disabling the creation and gc. Side effect: application cannot rollback. If you don't need to rollback applications, you can switch it on to reduce the storage and improve performance.") + flag.BoolVar(&workflow.DisableRecorder, "optimize-disable-workflow-recorder", false, "Optimize workflow recorder by disabling the creation and gc. Side effect: workflow will not record application after finished running. If you do not use VelaUX, you can switch it on to improve performance.") + flag.BoolVar(&wfContext.EnableInMemoryContext, "optimize-enable-in-memory-workflow-context", false, "Optimize workflow by use in-memory context. Side effect: controller crash will lead to workflow run again from scratch and possible to cause mistakes in workflow inputs/outputs. You can use this optimization when you don't use input/output feature of workflow.") + flag.BoolVar(&application.DisableResourceApplyDoubleCheck, "optimize-disable-resource-apply-double-check", false, "Optimize workflow by ignoring resource double check after apply. Side effect: controller will not wait for resource creation. If you want to use KubeVela to dispatch tons of resources and do not need to double check the creation result, you can enable this optimization.") + flag.BoolVar(&application.EnableResourceTrackerDeleteOnlyTrigger, "optimize-enable-resource-tracker-delete-only-trigger", true, "Optimize resourcetracker by only trigger reconcile when resourcetracker is deleted. It is enabled by default. If you want to integrate KubeVela with your own operator or allow ResourceTracker manual edit, you can turn it off.") +} diff --git a/pkg/monitor/context/context.go b/pkg/monitor/context/context.go index bcaff4c27..dabef3b32 100644 --- a/pkg/monitor/context/context.go +++ b/pkg/monitor/context/context.go @@ -85,7 +85,7 @@ func (t *traceContext) Commit(msg string) { msg = fmt.Sprintf("[Finished]: %s(%s)", t.id, msg) duration := time.Since(t.beginTimestamp) for _, export := range t.exporters { - export(t, duration.Microseconds()) + export(t, duration.Seconds()) } if t.logLevel == 0 { klog.InfoSDepth(1, msg, t.getTagsWith("duration", duration.String())...) @@ -163,11 +163,11 @@ func copySlice(in []interface{}) []interface{} { } // Exporter export context info. -type Exporter func(t *traceContext, duration int64) +type Exporter func(t *traceContext, duration float64) // DurationMetric export context duration metric. func DurationMetric(h func(v float64)) Exporter { - return func(t *traceContext, duration int64) { - h(float64(duration / 1000)) + return func(t *traceContext, duration float64) { + h(duration) } } diff --git a/pkg/monitor/metrics/application.go b/pkg/monitor/metrics/application.go new file mode 100644 index 000000000..10969acf0 --- /dev/null +++ b/pkg/monitor/metrics/application.go @@ -0,0 +1,132 @@ +/* + 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 metrics + +import ( + "github.com/prometheus/client_golang/prometheus" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +var ( + // CreateAppHandlerDurationHistogram report the create appHandler execution duration. + CreateAppHandlerDurationHistogram = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "create_app_handler_time_seconds", + Help: "create appHandler duration distributions, this operate will list ResourceTrackers.", + Buckets: histogramBuckets, + ConstLabels: prometheus.Labels{}, + }, []string{"controller"}) + + // HandleFinalizersDurationHistogram report the handle finalizers execution duration. + HandleFinalizersDurationHistogram = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "handle_finalizers_time_seconds", + Help: "handle finalizers duration distributions.", + Buckets: histogramBuckets, + ConstLabels: prometheus.Labels{}, + }, []string{"controller", "type"}) + + // ParseAppFileDurationHistogram report the parse appFile execution duration. + ParseAppFileDurationHistogram = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "parse_appFile_time_seconds", + Help: "parse appFile duration distributions.", + Buckets: histogramBuckets, + ConstLabels: prometheus.Labels{}, + }, []string{"controller"}) + + // PrepareCurrentAppRevisionDurationHistogram report the parse current appRevision execution duration. + PrepareCurrentAppRevisionDurationHistogram = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "prepare_current_appRevision_time_seconds", + Help: "parse current appRevision duration distributions.", + Buckets: histogramBuckets, + ConstLabels: prometheus.Labels{}, + }, []string{"controller"}) + + // ApplyAppRevisionDurationHistogram report the apply appRevision execution duration. + ApplyAppRevisionDurationHistogram = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "apply_appRevision_time_seconds", + Help: "apply appRevision duration distributions.", + Buckets: histogramBuckets, + ConstLabels: prometheus.Labels{}, + }, []string{"controller"}) + + // PrepareWorkflowAndPolicyDurationHistogram report the prepare workflow and policy execution duration. + PrepareWorkflowAndPolicyDurationHistogram = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "prepare_workflow_and_policy_time_seconds", + Help: "prepare workflow and policy duration distributions.", + Buckets: histogramBuckets, + ConstLabels: prometheus.Labels{}, + }, []string{"controller"}) + + // GCResourceTrackersDurationHistogram report the gc resourceTrackers execution duration. + GCResourceTrackersDurationHistogram = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "gc_resourceTrackers_time_seconds", + Help: "gc resourceTrackers duration distributions.", + Buckets: histogramBuckets, + ConstLabels: prometheus.Labels{}, + }, []string{"stage"}) + + // ClientRequestHistogram report the client request execution duration. + ClientRequestHistogram = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "client_request_time_seconds", + Help: "client request duration distributions.", + Buckets: histogramBuckets, + ConstLabels: prometheus.Labels{}, + }, []string{"verb", "Kind", "apiVersion", "unstructured", "cluster"}) + + // ApplicationReconcileTimeHistogram report the reconciling time cost of application controller with state transition recorded + ApplicationReconcileTimeHistogram = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "application_reconcile_time_seconds", + Help: "application reconcile duration distributions.", + Buckets: histogramBuckets, + ConstLabels: prometheus.Labels{}, + }, []string{"begin_phase", "end_phase"}) + + // ApplyComponentTimeHistogram report the time cost of applyComponentFunc + ApplyComponentTimeHistogram = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "apply_component_time_seconds", + Help: "apply component duration distributions.", + Buckets: histogramBuckets, + ConstLabels: prometheus.Labels{}, + }, []string{"stage"}) +) + +var ( + // ListResourceTrackerCounter report the list resource tracker number. + ListResourceTrackerCounter = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "list_resourcetracker_num", + Help: "list resourceTrackers times.", + }, []string{"controller"}) +) + +var ( + // ResourceTrackerNumberGauge report the number of resourceTracker + ResourceTrackerNumberGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Name: "resourcetracker_number", + Help: "resourceTracker number.", + }, []string{"application", "namespace"}) +) + +// 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/monitor/metrics/workflow.go b/pkg/monitor/metrics/workflow.go index cc56a042c..4861dc0ce 100644 --- a/pkg/monitor/metrics/workflow.go +++ b/pkg/monitor/metrics/workflow.go @@ -19,18 +19,39 @@ import ( "sigs.k8s.io/controller-runtime/pkg/metrics" ) +var histogramBuckets = []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.15, 0.2, 0.25, 0.3, 0.35, 0.4, 0.45, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0, + 1.25, 1.5, 1.75, 2.0, 2.5, 3.0, 3.5, 4.0, 4.5, 5, 6, 7, 8, 9, 10, 15, 20, 25, 30, 40, 50, 60} + var ( - // StepDurationSummary report the step execution duration summary. - StepDurationSummary = prometheus.NewSummaryVec(prometheus.SummaryOpts{ + // StepDurationHistogram report the step execution duration. + StepDurationHistogram = prometheus.NewHistogramVec(prometheus.HistogramOpts{ Name: "step_duration_ms", Help: "step latency distributions.", - Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001}, + Buckets: histogramBuckets, ConstLabels: prometheus.Labels{}, - }, []string{"application", "workflow_revision", "step_name", "step_type"}) + }, []string{"controller", "step_type"}) ) +var collectorGroup = []prometheus.Collector{ + CreateAppHandlerDurationHistogram, + HandleFinalizersDurationHistogram, + ParseAppFileDurationHistogram, + PrepareCurrentAppRevisionDurationHistogram, + ApplyAppRevisionDurationHistogram, + PrepareWorkflowAndPolicyDurationHistogram, + StepDurationHistogram, + GCResourceTrackersDurationHistogram, + ListResourceTrackerCounter, + ClientRequestHistogram, + ApplicationReconcileTimeHistogram, + ApplyComponentTimeHistogram, + ResourceTrackerNumberGauge, +} + func init() { - if err := metrics.Registry.Register(StepDurationSummary); err != nil { - klog.Error(err) + for _, collector := range collectorGroup { + if err := metrics.Registry.Register(collector); err != nil { + klog.Error(err) + } } } diff --git a/pkg/multicluster/utils.go b/pkg/multicluster/utils.go index 2d522e1b3..a46ce9e52 100644 --- a/pkg/multicluster/utils.go +++ b/pkg/multicluster/utils.go @@ -68,6 +68,12 @@ func ContextWithClusterName(ctx context.Context, clusterName string) context.Con return context.WithValue(ctx, ClusterContextKey, clusterName) } +// IsInLocalCluster check if target cluster is local cluster +func IsInLocalCluster(ctx context.Context) bool { + clusterName := ClusterNameInContext(ctx) + return clusterName == "" || clusterName == ClusterLocalName +} + // ContextInLocalCluster create context in local cluster func ContextInLocalCluster(ctx context.Context) context.Context { return context.WithValue(ctx, ClusterContextKey, ClusterLocalName) diff --git a/pkg/resourcekeeper/gc.go b/pkg/resourcekeeper/gc.go index b1172a5f6..45b94ff54 100644 --- a/pkg/resourcekeeper/gc.go +++ b/pkg/resourcekeeper/gc.go @@ -19,10 +19,12 @@ package resourcekeeper import ( "context" "encoding/json" + "math/rand" "strings" + "time" "github.com/crossplane/crossplane-runtime/pkg/meta" - version "github.com/hashicorp/go-version" + "github.com/hashicorp/go-version" "github.com/pkg/errors" v1 "k8s.io/api/apps/v1" kerrors "k8s.io/apimachinery/pkg/api/errors" @@ -32,12 +34,18 @@ import ( "github.com/oam-dev/kubevela/apis/core.oam.dev/v1alpha1" "github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1" + "github.com/oam-dev/kubevela/pkg/monitor/metrics" "github.com/oam-dev/kubevela/pkg/multicluster" "github.com/oam-dev/kubevela/pkg/oam" "github.com/oam-dev/kubevela/pkg/resourcetracker" version2 "github.com/oam-dev/kubevela/version" ) +var ( + // MarkWithProbability optimize ResourceTracker gc for legacy resource by reducing the frequency of outdated rt check + MarkWithProbability = 0.1 +) + // GCOption option for gc type GCOption interface { ApplyToGCConfig(*gcConfig) @@ -134,7 +142,17 @@ type gcHandler struct { cfg *gcConfig } +func (h *gcHandler) monitor(stage string) func() { + begin := time.Now() + return func() { + v := time.Since(begin).Seconds() + metrics.GCResourceTrackersDurationHistogram.WithLabelValues(stage).Observe(v) + } +} + func (h *gcHandler) Init() { + cb := h.monitor("init") + defer cb() h.cache.registerResourceTrackers(append(h._historyRTs, h._currentRT, h._rootRT)...) } @@ -145,6 +163,9 @@ func (h *gcHandler) scan(ctx context.Context) (inactiveRTs []*v1beta1.ResourceTr } else { if h.cfg.passive { inactiveRTs = []*v1beta1.ResourceTracker{} + if rand.Float64() > MarkWithProbability { //nolint + return inactiveRTs + } for _, rt := range h._historyRTs { if rt != nil { inactive := true @@ -168,6 +189,8 @@ func (h *gcHandler) scan(ctx context.Context) (inactiveRTs []*v1beta1.ResourceTr } func (h *gcHandler) Mark(ctx context.Context) error { + cb := h.monitor("mark") + defer cb() inactiveRTs := h.scan(ctx) for _, rt := range inactiveRTs { if rt != nil && rt.GetDeletionTimestamp() == nil { @@ -203,6 +226,8 @@ func (h *gcHandler) checkAndRemoveResourceTrackerFinalizer(ctx context.Context, } func (h *gcHandler) Sweep(ctx context.Context) (finished bool, waiting []v1beta1.ManagedResource, err error) { + cb := h.monitor("sweep") + defer cb() finished = true for _, rt := range append(h._historyRTs, h._currentRT, h._rootRT) { if rt != nil && rt.GetDeletionTimestamp() != nil { @@ -238,6 +263,8 @@ func (h *gcHandler) recycleResourceTracker(ctx context.Context, rt *v1beta1.Reso } func (h *gcHandler) Finalize(ctx context.Context) error { + cb := h.monitor("finalize") + defer cb() for _, rt := range append(h._historyRTs, h._currentRT, h._rootRT) { if rt != nil && rt.GetDeletionTimestamp() != nil && meta.FinalizerExists(rt, resourcetracker.Finalizer) { if err := h.recycleResourceTracker(ctx, rt); err != nil { @@ -249,6 +276,8 @@ func (h *gcHandler) Finalize(ctx context.Context) error { } func (h *gcHandler) GarbageCollectComponentRevisionResourceTracker(ctx context.Context) error { + cb := h.monitor("comp-rev") + defer cb() if h._crRT == nil { return nil } @@ -326,7 +355,7 @@ func (h *gcHandler) GarbageCollectLegacyResourceTrackers(ctx context.Context) er for cluster := range clusters { _ctx := multicluster.ContextWithClusterName(ctx, cluster) rts := &unstructured.UnstructuredList{} - rts.SetGroupVersionKind(v1beta1.SchemeGroupVersion.WithKind("ResourceTracker")) + rts.SetGroupVersionKind(v1beta1.SchemeGroupVersion.WithKind("ResourceTrackerList")) if err = h.Client.List(_ctx, rts, client.MatchingLabels(map[string]string{ oam.LabelAppName: h.app.Name, oam.LabelAppNamespace: h.app.Namespace, diff --git a/pkg/resourcekeeper/gc_test.go b/pkg/resourcekeeper/gc_test.go index 50d73086f..0f9351598 100644 --- a/pkg/resourcekeeper/gc_test.go +++ b/pkg/resourcekeeper/gc_test.go @@ -38,6 +38,7 @@ import ( ) func TestResourceKeeperGarbageCollect(t *testing.T) { + MarkWithProbability = 1.0 r := require.New(t) cli := fake.NewClientBuilder().WithScheme(common.Scheme).Build() ctx := context.Background() diff --git a/pkg/resourcetracker/app.go b/pkg/resourcetracker/app.go index 1bab77483..c9b84f1c8 100644 --- a/pkg/resourcetracker/app.go +++ b/pkg/resourcetracker/app.go @@ -20,6 +20,8 @@ import ( "context" "fmt" + "github.com/oam-dev/kubevela/pkg/monitor/metrics" + "github.com/crossplane/crossplane-runtime/pkg/meta" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "sigs.k8s.io/controller-runtime/pkg/client" @@ -99,6 +101,7 @@ func CreateComponentRevisionResourceTracker(ctx context.Context, cli client.Clie // ListApplicationResourceTrackers list resource trackers for application with all historyRTs sorted by version number func ListApplicationResourceTrackers(ctx context.Context, cli client.Client, app *v1beta1.Application) (rootRT *v1beta1.ResourceTracker, currentRT *v1beta1.ResourceTracker, historyRTs []*v1beta1.ResourceTracker, crRT *v1beta1.ResourceTracker, err error) { + metrics.ListResourceTrackerCounter.WithLabelValues("application").Inc() rts := v1beta1.ResourceTrackerList{} if err = cli.List(ctx, &rts, client.MatchingLabels{ oam.LabelAppName: app.Name, @@ -141,12 +144,16 @@ func ListApplicationResourceTrackers(ctx context.Context, cli client.Client, app // RecordManifestInResourceTracker records resources in ResourceTracker func RecordManifestInResourceTracker(ctx context.Context, cli client.Client, rt *v1beta1.ResourceTracker, manifest *unstructured.Unstructured, metaOnly bool) error { - rt.AddManagedResource(manifest, metaOnly) + if updated := rt.AddManagedResource(manifest, metaOnly); !updated { + return nil + } return cli.Update(ctx, rt) } // DeletedManifestInResourceTracker marks resources as deleted in resourcetracker, if remove is true, resources will be removed from resourcetracker func DeletedManifestInResourceTracker(ctx context.Context, cli client.Client, rt *v1beta1.ResourceTracker, manifest *unstructured.Unstructured, remove bool) error { - rt.DeleteManagedResource(manifest, remove) + if updated := rt.DeleteManagedResource(manifest, remove); !updated { + return nil + } return cli.Update(ctx, rt) } diff --git a/pkg/resourcetracker/optimize.go b/pkg/resourcetracker/optimize.go new file mode 100644 index 000000000..4680b50b2 --- /dev/null +++ b/pkg/resourcetracker/optimize.go @@ -0,0 +1,69 @@ +/* +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 resourcetracker + +import ( + "context" + + "sigs.k8s.io/controller-runtime/pkg/cache" + "sigs.k8s.io/controller-runtime/pkg/client" + + "github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1" + "github.com/oam-dev/kubevela/pkg/oam" +) + +// appIndex identify the index for resourcetracker to accelerate cache retrieval +const appIndex = "app" + +var ( + // OptimizeListOp optimize ResourceTracker List Op by adding index + OptimizeListOp = true +) + +// ExtendResourceTrackerListOption wraps list rt options by adding indexing fields +func ExtendResourceTrackerListOption(list client.ObjectList, opts []client.ListOption) []client.ListOption { + if !OptimizeListOp { + return opts + } + if _, ok := list.(*v1beta1.ResourceTrackerList); ok { + for _, opt := range opts { + if ml, isML := opt.(client.MatchingLabels); isML { + appName := ml[oam.LabelAppName] + appNs := ml[oam.LabelAppNamespace] + if appName != "" { + opts = append(opts, client.MatchingFields(map[string]string{ + appIndex: appNs + "/" + appName, + })) + } + } + } + } + return opts +} + +// AddResourceTrackerCacheIndex add indexing configuration for cache +func AddResourceTrackerCacheIndex(cache cache.Cache) error { + if !OptimizeListOp { + return nil + } + return cache.IndexField(context.Background(), &v1beta1.ResourceTracker{}, appIndex, func(obj client.Object) []string { + if labels := obj.GetLabels(); labels != nil { + return []string{labels[oam.LabelAppNamespace] + "/" + labels[oam.LabelAppName]} + } + return []string{} + }) +} diff --git a/pkg/workflow/context/context.go b/pkg/workflow/context/context.go index e1f44dbb4..def2331ba 100644 --- a/pkg/workflow/context/context.go +++ b/pkg/workflow/context/context.go @@ -199,7 +199,9 @@ func (wf *WorkflowContext) writeToStore() error { func (wf *WorkflowContext) sync() error { ctx := context.Background() - if err := wf.cli.Update(ctx, wf.store); err != nil { + if EnableInMemoryContext { + MemStore.UpdateInMemoryContext(wf.store) + } else if err := wf.cli.Update(ctx, wf.store); err != nil { if kerrors.IsNotFound(err) { return wf.cli.Create(ctx, wf.store) } @@ -331,7 +333,9 @@ func newContext(cli client.Client, ns, app string, appUID types.UID) (*WorkflowC Controller: pointer.BoolPtr(true), }, }) - if err := cli.Get(ctx, client.ObjectKey{Name: store.Name, Namespace: store.Namespace}, &store); err != nil { + if EnableInMemoryContext { + MemStore.GetOrCreateInMemoryContext(&store) + } else if err := cli.Get(ctx, client.ObjectKey{Name: store.Name, Namespace: store.Namespace}, &store); err != nil { if kerrors.IsNotFound(err) { if err := cli.Create(ctx, &store); err != nil { return nil, err @@ -358,7 +362,11 @@ func newContext(cli client.Client, ns, app string, appUID types.UID) (*WorkflowC // LoadContext load workflow context from store. func LoadContext(cli client.Client, ns, app string) (Context, error) { var store corev1.ConfigMap - if err := cli.Get(context.Background(), client.ObjectKey{ + store.Name = generateStoreName(app) + store.Namespace = ns + if EnableInMemoryContext { + MemStore.GetOrCreateInMemoryContext(&store) + } else if err := cli.Get(context.Background(), client.ObjectKey{ Namespace: ns, Name: generateStoreName(app), }, &store); err != nil { diff --git a/pkg/workflow/context/storage.go b/pkg/workflow/context/storage.go new file mode 100644 index 000000000..273d4954f --- /dev/null +++ b/pkg/workflow/context/storage.go @@ -0,0 +1,80 @@ +/* +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 context + +import ( + "fmt" + "sync" + + v1 "k8s.io/api/core/v1" +) + +var ( + // EnableInMemoryContext optimize workflow context storage by storing it in memory instead of etcd + EnableInMemoryContext = false +) + +type inMemoryContextStorage struct { + mu sync.Mutex + contexts map[string]*v1.ConfigMap +} + +// MemStore store in-memory context +var MemStore = &inMemoryContextStorage{ + contexts: map[string]*v1.ConfigMap{}, +} + +func (o *inMemoryContextStorage) getKey(cm *v1.ConfigMap) string { + ns := cm.GetNamespace() + if ns == "" { + ns = "default" + } + name := cm.GetName() + return ns + "/" + name +} + +func (o *inMemoryContextStorage) GetOrCreateInMemoryContext(cm *v1.ConfigMap) { + if obj := o.GetInMemoryContext(cm.Name, cm.Namespace); obj != nil { + obj.DeepCopyInto(cm) + } else { + o.CreateInMemoryContext(cm) + } +} + +func (o *inMemoryContextStorage) GetInMemoryContext(name, ns string) *v1.ConfigMap { + return o.contexts[ns+"/"+name] +} + +func (o *inMemoryContextStorage) CreateInMemoryContext(cm *v1.ConfigMap) { + o.mu.Lock() + defer o.mu.Unlock() + cm.Data = map[string]string{} + o.contexts[o.getKey(cm)] = cm +} + +func (o *inMemoryContextStorage) UpdateInMemoryContext(cm *v1.ConfigMap) { + o.mu.Lock() + defer o.mu.Unlock() + o.contexts[o.getKey(cm)] = cm +} + +func (o *inMemoryContextStorage) DeleteInMemoryContext(appName string) { + o.mu.Lock() + defer o.mu.Unlock() + key := fmt.Sprintf("workflow-%s-context", appName) + delete(o.contexts, key) +} diff --git a/pkg/workflow/workflow.go b/pkg/workflow/workflow.go index ff9e34341..67f138f0b 100644 --- a/pkg/workflow/workflow.go +++ b/pkg/workflow/workflow.go @@ -42,6 +42,11 @@ import ( wfTypes "github.com/oam-dev/kubevela/pkg/workflow/types" ) +var ( + // DisableRecorder optimize workflow by disable recorder + DisableRecorder = false +) + const ( // minWorkflowBackoffWaitTime is the min time to wait before reconcile workflow again minWorkflowBackoffWaitTime = 1 @@ -179,6 +184,9 @@ func (w *workflow) ExecuteSteps(ctx monitorContext.Context, appRev *oamcore.Appl // Trace record the workflow execute history. func (w *workflow) Trace() error { + if DisableRecorder { + return nil + } data, err := json.Marshal(w.app) if err != nil { return err @@ -440,7 +448,7 @@ func (e *engine) steps(taskRunners []wfTypes.TaskRunner) error { status, operation, err := runner.Run(wfCtx, &wfTypes.TaskRunOptions{ GetTracer: func(id string, stepStatus oamcore.WorkflowStep) monitorContext.Context { return e.monitorCtx.Fork(id, monitorContext.DurationMetric(func(v float64) { - metrics.StepDurationSummary.WithLabelValues(e.app.Namespace+"/"+e.app.Name, e.status.AppRevision, stepStatus.Name, stepStatus.Type).Observe(v) + metrics.StepDurationHistogram.WithLabelValues("application", stepStatus.Type).Observe(v) })) }, })