fix: memory leak

cmd/agent/args/args.go

@@ -48,6 +48,8 @@ const (
 
 	defaultProfilerPath    = "/debug/pprof/"
 	defaultProfilerAddress = ":7777"
+
+	defaultPyroscopeAddress = "http://pyroscope.monitoring.svc.cluster.local:4040"
 )
 
 var (
@@ -56,6 +58,7 @@ var (
 	argEnableLeaderElection            = flag.Bool("leader-elect", false, "Enable leader election for controller manager. Enabling this will ensure there is only one active controller manager.")
 	argLocal                           = flag.Bool("local", false, "Whether you're running the operator locally.")
 	argProfiler                        = flag.Bool("profiler", false, "Enable pprof handler. By default it will be exposed on localhost:7777 under '/debug/pprof'")
+	argPyroscope                       = flag.Bool("pyroscope", true, "Enable pyroscope integration for detailed application profiling. By default it will push to http://pyroscope.monitoring.svc.cluster.local:4040")
 	argDisableResourceCache            = flag.Bool("disable-resource-cache", false, "Control whether resource cache should be enabled or not.")
 	argEnableKubecostProxy             = flag.Bool("enable-kubecost-proxy", false, "If set, will proxy a Kubecost API request through the K8s API server.")
 
@@ -77,6 +80,7 @@ var (
 	argControllerCacheTTL = flag.String("controller-cache-ttl", defaultControllerCacheTTL, "The time to live of console controller cache entries.")
 	argRestoreNamespace   = flag.String("restore-namespace", defaultRestoreNamespace, "The namespace where Velero restores are located.")
 	argServices           = flag.String("services", "", "A comma separated list of service ids to reconcile. Leave empty to reconcile all.")
+	argPyroscopeAddress   = flag.String("pyroscope-address", defaultPyroscopeAddress, "The address of the Pyroscope server.")
 
 	serviceSet containers.Set[string]
 )
@@ -268,6 +272,14 @@ func ResourceCacheEnabled() bool {
 	return !(*argDisableResourceCache)
 }
 
+func PyroscopeEnabled() bool {
+	return *argPyroscope
+}
+
+func PyroscopeAddress() string {
+	return *argPyroscopeAddress
+}
+
 func ensureOrDie(argName string, arg *string) {
 	if arg == nil || len(*arg) == 0 {
 		pflag.PrintDefaults()

cmd/agent/args/pyroscope.go

@@ -0,0 +1,49 @@
+package args
+
+import (
+	"os"
+	"runtime"
+	"time"
+
+	"github.com/grafana/pyroscope-go"
+	"k8s.io/klog/v2"
+)
+
+func InitPyroscope() (*pyroscope.Profiler, error) {
+	klog.Info("initializing pyroscope")
+
+	runtime.SetMutexProfileFraction(5)
+	runtime.SetBlockProfileRate(5)
+
+	return pyroscope.Start(pyroscope.Config{
+		ApplicationName: "deployment-operator",
+
+		// replace this with the address of pyroscope server
+		ServerAddress: PyroscopeAddress(),
+
+		// you can disable logging by setting this to nil
+		Logger: nil,
+
+		// Push metrics once every 30 seconds
+		UploadRate: 30 * time.Second,
+
+		// you can provide static tags via a map:
+		Tags: map[string]string{"hostname": os.Getenv("HOSTNAME")},
+
+		ProfileTypes: []pyroscope.ProfileType{
+			// these profile types are enabled by default:
+			pyroscope.ProfileCPU,
+			pyroscope.ProfileAllocObjects,
+			pyroscope.ProfileAllocSpace,
+			pyroscope.ProfileInuseObjects,
+			pyroscope.ProfileInuseSpace,
+
+			// these profile types are optional:
+			pyroscope.ProfileGoroutines,
+			pyroscope.ProfileMutexCount,
+			pyroscope.ProfileMutexDuration,
+			pyroscope.ProfileBlockCount,
+			pyroscope.ProfileBlockDuration,
+		},
+	})
+}

cmd/agent/main.go

@@ -10,12 +10,6 @@ import (
 	certmanagerv1 "github.com/cert-manager/cert-manager/pkg/apis/certmanager/v1"
 	templatesv1 "github.com/open-policy-agent/frameworks/constraint/pkg/apis/templates/v1"
 	constraintstatusv1beta1 "github.com/open-policy-agent/gatekeeper/v3/apis/status/v1beta1"
-	deploymentsv1alpha1 "github.com/pluralsh/deployment-operator/api/v1alpha1"
-	"github.com/pluralsh/deployment-operator/cmd/agent/args"
-	"github.com/pluralsh/deployment-operator/pkg/cache"
-	"github.com/pluralsh/deployment-operator/pkg/client"
-	consolectrl "github.com/pluralsh/deployment-operator/pkg/controller"
-	"github.com/pluralsh/deployment-operator/pkg/scraper"
 	velerov1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -25,6 +19,13 @@ import (
 	"k8s.io/client-go/rest"
 	"k8s.io/klog/v2"
 	ctrl "sigs.k8s.io/controller-runtime"
+
+	deploymentsv1alpha1 "github.com/pluralsh/deployment-operator/api/v1alpha1"
+	"github.com/pluralsh/deployment-operator/cmd/agent/args"
+	"github.com/pluralsh/deployment-operator/pkg/cache"
+	"github.com/pluralsh/deployment-operator/pkg/client"
+	consolectrl "github.com/pluralsh/deployment-operator/pkg/controller"
+	"github.com/pluralsh/deployment-operator/pkg/scraper"
 )
 
 var (
@@ -52,8 +53,23 @@ const (
 func main() {
 	args.Init()
 	config := ctrl.GetConfigOrDie()
+	// Use protobuf instead of application/json
+	config.ContentType = "application/vnd.kubernetes.protobuf"
+	config.AcceptContentTypes = "application/vnd.kubernetes.protobuf,application/json"
 	ctx := ctrl.LoggerInto(ctrl.SetupSignalHandler(), setupLog)
 
+	if args.PyroscopeEnabled() {
+		profiler, err := args.InitPyroscope()
+		if err != nil {
+			setupLog.Error(err, "unable to initialize pyroscope")
+			os.Exit(1)
+		}
+
+		defer func() {
+			_ = profiler.Stop()
+		}()
+	}
+
 	extConsoleClient := client.New(args.ConsoleUrl(), args.DeployToken())
 	discoveryClient := initDiscoveryClientOrDie(config)
 	kubeManager := initKubeManagerOrDie(config)

go.mod

@@ -247,6 +247,8 @@ require (
 	github.com/gostaticanalysis/forcetypeassert v0.1.0 // indirect
 	github.com/gostaticanalysis/nilerr v0.1.1 // indirect
 	github.com/gosuri/uitable v0.0.4 // indirect
+	github.com/grafana/pyroscope-go v1.2.0 // indirect
+	github.com/grafana/pyroscope-go/godeltaprof v0.1.8 // indirect
 	github.com/gregjones/httpcache v0.0.0-20190611155906-901d90724c79 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-cleanhttp v0.5.2 // indirect

go.sum

@@ -874,6 +874,10 @@ github.com/gostaticanalysis/testutil v0.4.0 h1:nhdCmubdmDF6VEatUNjgUZBJKWRqugoIS
 github.com/gostaticanalysis/testutil v0.4.0/go.mod h1:bLIoPefWXrRi/ssLFWX1dx7Repi5x3CuviD3dgAZaBU=
 github.com/gosuri/uitable v0.0.4 h1:IG2xLKRvErL3uhY6e1BylFzG+aJiwQviDDTfOKeKTpY=
 github.com/gosuri/uitable v0.0.4/go.mod h1:tKR86bXuXPZazfOTG1FIzvjIdXzd0mo4Vtn16vt0PJo=
+github.com/grafana/pyroscope-go v1.2.0 h1:aILLKjTj8CS8f/24OPMGPewQSYlhmdQMBmol1d3KGj8=
+github.com/grafana/pyroscope-go v1.2.0/go.mod h1:2GHr28Nr05bg2pElS+dDsc98f3JTUh2f6Fz1hWXrqwk=
+github.com/grafana/pyroscope-go/godeltaprof v0.1.8 h1:iwOtYXeeVSAeYefJNaxDytgjKtUuKQbJqgAIjlnicKg=
+github.com/grafana/pyroscope-go/godeltaprof v0.1.8/go.mod h1:2+l7K7twW49Ct4wFluZD3tZ6e0SjanjcUUBPVD/UuGU=
 github.com/gregjones/httpcache v0.0.0-20190611155906-901d90724c79 h1:+ngKgrYPPJrOjhax5N+uePQ0Fh1Z7PheYoUI/0nzkPA=
 github.com/gregjones/httpcache v0.0.0-20190611155906-901d90724c79/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 h1:Ovs26xHkKqVztRpIrF/92BcuyuQ/YW4NSIpoGtfXNho=

internal/kstatus/watcher/object_status_reporter.go

@@ -15,7 +15,6 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apimachinery/pkg/watch"
@@ -320,17 +319,13 @@ func (in *ObjectStatusReporter) startInformerWithRetry(ctx context.Context, gkn
 	retryCount := 0
 
 	wait.BackoffUntil(func() {
-		err := in.startInformerNow(
-			ctx,
-			gkn,
-		)
-		if retryCount >= 10 {
+		if retryCount >= 2 {
 			klog.V(3).Infof("Watch start abort, reached retry limit: %d: %v", retryCount, gkn)
 			in.stopInformer(gkn)
 			return
 		}
 
-		if err != nil {
+		if err := in.startInformerNow(ctx, gkn); err != nil {
 			if meta.IsNoMatchError(err) {
 				// CRD (or api extension) not installed, retry
 				klog.V(3).Infof("Watch start error (blocking until CRD is added): %v: %v", gkn, err)
@@ -353,6 +348,7 @@ func (in *ObjectStatusReporter) startInformerWithRetry(ctx context.Context, gkn
 			in.handleFatalError(eventCh, err)
 			return
 		}
+
 		// Success! - Stop retrying
 		retryCancel()
 	}, backoffManager, true, retryCtx.Done())
@@ -372,17 +368,27 @@ func (in *ObjectStatusReporter) newWatcher(ctx context.Context, gkn GroupKindNam
 		labelSelectorString = in.LabelSelector.String()
 	}
 
-	w, err := watcher.NewRetryListerWatcher(
-		watcher.WithListWatchFunc(
-			func(options metav1.ListOptions) (runtime.Object, error) {
-				options.LabelSelector = labelSelectorString
-				return in.DynamicClient.Resource(gvr).List(ctx, options)
-			}, func(options metav1.ListOptions) (watch.Interface, error) {
-				options.LabelSelector = labelSelectorString
-				return in.DynamicClient.Resource(gvr).Watch(ctx, options)
-			}),
-		watcher.WithID(gkn.String()),
-	)
+	//w, err := watcher.NewRetryListerWatcher(
+	//	ctx,
+	//	watcher.WithListWatchFunc(
+	//		func(options metav1.ListOptions) (runtime.Object, error) {
+	//			options.LabelSelector = labelSelectorString
+	//			return in.DynamicClient.Resource(gvr).List(ctx, options)
+	//		}, func(options metav1.ListOptions) (watch.Interface, error) {
+	//			options.LabelSelector = labelSelectorString
+	//			return in.DynamicClient.Resource(gvr).Watch(ctx, options)
+	//		}),
+	//	watcher.WithID(fmt.Sprintf("%s/%s", in.id, gkn.String())),
+	//)
+
+	w, err := watcher.NewRetryWatcher("", &cache.ListWatch{
+		WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
+			options.LabelSelector = labelSelectorString
+			return in.DynamicClient.Resource(gvr).Watch(ctx, options)
+		},
+		DisableChunking: true,
+	})
+
 	if apierrors.IsNotFound(err) && !in.hasGVR(gvr) {
 		return nil, &meta.NoKindMatchError{
 			GroupKind:        gk,
@@ -458,7 +464,7 @@ func (in *ObjectStatusReporter) Run(stopCh <-chan struct{}, echan <-chan watch.E
 			return
 		case e, ok := <-echan:
 			if !ok {
-				klog.Error("event channel closed")
+				klog.V(4).Info("event channel closed")
 				return
 			}