From bdf71bb2908085aa4e31af596726aa28720d2b35 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Fri, 23 Sep 2022 17:14:10 +0800 Subject: [PATCH] [Backport release-1.5] Fix: memory leak of the apiserver (#4777) * Fix: memory leak of the apiserver Signed-off-by: barnettZQG (cherry picked from commit 0a8a70730f5852580175a91b4cc356911f1ad2ce) * Fix: listen to the context done event Signed-off-by: barnettZQG (cherry picked from commit dfb81224cbf71b8f871f4999ae29977f330fa925) * Fix: remove the shutdown code Signed-off-by: barnettZQG (cherry picked from commit a331b2c54abfd1c89d54f87860d96a24414cc71c) Co-authored-by: barnettZQG --- cmd/apiserver/main.go | 7 ++++ cmd/core/main.go | 34 ++------------------ pkg/apiserver/config/config.go | 3 ++ pkg/apiserver/event/sync/worker.go | 19 +++++++---- pkg/utils/pprof.go | 51 ++++++++++++++++++++++++++++++ 5 files changed, 75 insertions(+), 39 deletions(-) create mode 100644 pkg/utils/pprof.go diff --git a/cmd/apiserver/main.go b/cmd/apiserver/main.go index 6567bdee9..aaf983ed4 100644 --- a/cmd/apiserver/main.go +++ b/cmd/apiserver/main.go @@ -35,6 +35,7 @@ import ( "github.com/oam-dev/kubevela/pkg/apiserver/config" "github.com/oam-dev/kubevela/pkg/apiserver/utils/log" "github.com/oam-dev/kubevela/pkg/features" + "github.com/oam-dev/kubevela/pkg/utils" "github.com/oam-dev/kubevela/version" ) @@ -50,6 +51,7 @@ func main() { flag.DurationVar(&s.serverConfig.LeaderConfig.Duration, "duration", time.Second*5, "the lease lock resource name") flag.DurationVar(&s.serverConfig.AddonCacheTime, "addon-cache-duration", time.Minute*10, "how long between two addon cache operation") flag.BoolVar(&s.serverConfig.DisableStatisticCronJob, "disable-statistic-cronJob", false, "close the system statistic info calculating cronJob") + flag.StringVar(&s.serverConfig.PprofAddr, "pprof-addr", "", "The address for pprof to use while exporting profiling results. The default value is empty which means do not expose it. Set it to address like :6666 to expose it.") flag.Float64Var(&s.serverConfig.KubeQPS, "kube-api-qps", 100, "the qps for kube clients. Low qps may lead to low throughput. High qps may give stress to api-server.") flag.IntVar(&s.serverConfig.KubeBurst, "kube-api-burst", 300, "the burst for kube clients. Recommend setting it qps*3.") features.APIServerMutableFeatureGate.AddFlag(flag.CommandLine) @@ -90,6 +92,11 @@ func main() { errChan := make(chan error) ctx, cancel := context.WithCancel(context.Background()) defer cancel() + + if s.serverConfig.PprofAddr != "" { + go utils.EnablePprof(s.serverConfig.PprofAddr, errChan) + } + go func() { if err := s.run(ctx, errChan); err != nil { errChan <- fmt.Errorf("failed to run apiserver: %w", err) diff --git a/cmd/core/main.go b/cmd/core/main.go index 41388c88d..2a66a953d 100644 --- a/cmd/core/main.go +++ b/cmd/core/main.go @@ -22,8 +22,6 @@ import ( goflag "flag" "fmt" "io" - "net/http" - "net/http/pprof" "os" "path/filepath" "strconv" @@ -53,6 +51,7 @@ import ( "github.com/oam-dev/kubevela/pkg/oam" "github.com/oam-dev/kubevela/pkg/oam/discoverymapper" "github.com/oam-dev/kubevela/pkg/resourcekeeper" + pkgutils "github.com/oam-dev/kubevela/pkg/utils" "github.com/oam-dev/kubevela/pkg/utils/common" "github.com/oam-dev/kubevela/pkg/utils/system" "github.com/oam-dev/kubevela/pkg/utils/util" @@ -159,36 +158,7 @@ func main() { if pprofAddr != "" { // Start pprof server if enabled - mux := http.NewServeMux() - mux.HandleFunc("/debug/pprof/", pprof.Index) - mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline) - mux.HandleFunc("/debug/pprof/profile", pprof.Profile) - mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol) - mux.HandleFunc("/debug/pprof/trace", pprof.Trace) - pprofServer := http.Server{ - Addr: pprofAddr, - Handler: mux, - } - klog.InfoS("Starting debug HTTP server", "addr", pprofServer.Addr) - - go func() { - go func() { - ctx := context.Background() - <-ctx.Done() - - ctx, cancelFunc := context.WithTimeout(context.Background(), 60*time.Minute) - defer cancelFunc() - - if err := pprofServer.Shutdown(ctx); err != nil { - klog.Error(err, "Failed to shutdown debug HTTP server") - } - }() - - if err := pprofServer.ListenAndServe(); !errors.Is(http.ErrServerClosed, err) { - klog.Error(err, "Failed to start debug HTTP server") - panic(err) - } - }() + go pkgutils.EnablePprof(pprofAddr, nil) } if logFilePath != "" { diff --git a/pkg/apiserver/config/config.go b/pkg/apiserver/config/config.go index d5cb03d18..6af000736 100644 --- a/pkg/apiserver/config/config.go +++ b/pkg/apiserver/config/config.go @@ -46,6 +46,9 @@ type Config struct { // KubeQPS the QPS of kube client KubeQPS float64 + + // PprofAddr the address for pprof to use while exporting profiling results. + PprofAddr string } type leaderConfig struct { diff --git a/pkg/apiserver/event/sync/worker.go b/pkg/apiserver/event/sync/worker.go index b356582b2..1ca62467f 100644 --- a/pkg/apiserver/event/sync/worker.go +++ b/pkg/apiserver/event/sync/worker.go @@ -18,11 +18,12 @@ package sync import ( "context" - "encoding/json" "sync" "github.com/fatih/color" v1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime" "k8s.io/client-go/dynamic" dynamicInformer "k8s.io/client-go/dynamic/dynamicinformer" "k8s.io/client-go/rest" @@ -58,14 +59,17 @@ func (a *ApplicationSync) Start(ctx context.Context, errorChan chan error) { factory := dynamicInformer.NewFilteredDynamicSharedInformerFactory(dynamicClient, 0, v1.NamespaceAll, nil) informer := factory.ForResource(v1beta1.SchemeGroupVersion.WithResource("applications")).Informer() getApp := func(obj interface{}) *v1beta1.Application { - app := &v1beta1.Application{} - bs, err := json.Marshal(obj) - if err != nil { - log.Logger.Errorf("decode the application failure %s", err.Error()) + if app, ok := obj.(*v1beta1.Application); ok { return app } - _ = json.Unmarshal(bs, app) - return app + var app v1beta1.Application + if object, ok := obj.(*unstructured.Unstructured); ok { + if err := runtime.DefaultUnstructuredConverter.FromUnstructured(object.Object, &app); err != nil { + log.Logger.Errorf("decode the application failure %s", err.Error()) + return &app + } + } + return &app } cu := &CR2UX{ ds: a.Store, @@ -89,6 +93,7 @@ func (a *ApplicationSync) Start(ctx context.Context, errorChan chan error) { if err := cu.AddOrUpdate(ctx, app.(*v1beta1.Application)); err != nil { log.Logger.Errorf("fail to add or update application %s", err.Error()) } + a.Queue.Done(app) } }() diff --git a/pkg/utils/pprof.go b/pkg/utils/pprof.go new file mode 100644 index 000000000..9a433f6eb --- /dev/null +++ b/pkg/utils/pprof.go @@ -0,0 +1,51 @@ +/* +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 utils + +import ( + "net/http" + "net/http/pprof" + + "k8s.io/klog/v2" +) + +// EnablePprof listen to the pprofAddr and export the profiling results +// If the errChan is nil, this function will panic when the listening error occurred. +func EnablePprof(pprofAddr string, errChan chan error) { + // Start pprof server if enabled + mux := http.NewServeMux() + mux.HandleFunc("/debug/pprof/", pprof.Index) + mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline) + mux.HandleFunc("/debug/pprof/profile", pprof.Profile) + mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol) + mux.HandleFunc("/debug/pprof/trace", pprof.Trace) + pprofServer := http.Server{ + Addr: pprofAddr, + Handler: mux, + } + + klog.InfoS("Starting debug HTTP server", "addr", pprofServer.Addr) + + if err := pprofServer.ListenAndServe(); err != nil { + klog.Error(err, "Failed to start debug HTTP server") + if errChan != nil { + errChan <- err + } else { + panic(err) + } + } +}