fix(agent): memory leaks and optimize resource usage with Datadog/Pyroscope integrations (#351)

* an attempt to fix memory leak

* add pyroscope profiler support

* use protobuf instead of application/json for api access and try to avoid retaining list items

* fallback to json in case protobuf is not accepted

* update pyroscope config

* fix lint

* use single content type for k8s config

* improve retry watcher list and k8s config

* try using watch with initial list without dedicated list

* start with empty resource version to get synthetic list first

* revert retry watcher changes

* add some more grafana metrics and decrease watcher restart attempts

* disable RetryListerWatcher and use raw RetryWatcher

* revert parts of the objeststatusreporter logic and use original reporter for cli utils

* `revert some changes and start from scratch`

* `revert some more code`

* `refactor: enhance RetryListerWatcher with stop logic and improve watcher handling while cleaning up unused metrics`

* refactor: fix lint issues, clean up imports, and improve cache and reconciliation logic

* fix: add buffer size to resultChan in RetryListerWatcher to prevent potential blocking

* feat: add Datadog profiling integration to deployment-operator agent

* chore(dependencies): update go module dependencies in go.sum

* feat: update Datadog profiler to support custom agent address configuration

* `feat(agent): add Datadog tracer initialization and refactor Datadog address to host configuration`

* feat: add service name and environment configuration to Datadog tracer setup

* `chore: update go module dependencies to the latest compatible versions`

* refactor: update Datadog tracer and profiler initialization with improved address handling and unified configuration

* fix(datadog): correct argument assignment for tracer setup in deployment-operator agent configuration

* fix(datadog): correct argument assignment for tracer setup in deployment-operator agent configuration

* feat: add support for configurable Datadog environment with new `datadog-env` flag

* feat: integrate Kubernetes client tracing with DataDog and update dependencies

* `refactor: move Kubernetes client trace setup closer to initialization for improved code clarity`

* `fix: update cache expiration logic to refresh timestamps and retain entries with cleared specific fields`

* `feat: add tracing to RetryListerWatcher for improved observability`

* feat(retry-watcher): add context support, enhance synchronization, and improve cleanup handling

* `ci: update golangci-lint-action to v7.0.0 and linter version to v2.3.4`

* ci: downgrade golangci-lint-action to v2.1.2 in workflow configuration

* fix(args): change default value of Datadog integration flag to false

* `test: fix test initialization by moving Init call to BeforeAll setup`

* fix(watcher): improve stop handling and add tracing in RetryListerWatcher

* feat: add profiling period and CPU duration configuration to Datadog profiler setup

* add prealloc linter

* fix prealloc issues

* add more linters

* `fix: use thread-safe check for watcher started status`

* test: move Init invocation to specific test cases in resource_cache_test

* feat: add environment variable support for feature toggles in deployment operator flags

* add more linters

* refactor(agent): remove protobuf content type configuration from agent main setup

* refactor: remove Datadog tracing spans from RetryListerWatcher implementation

---------

Co-authored-by: Marcin Maciaszczyk <[email protected]>

Author: floreks

.github/workflows/ci.yaml

@@ -40,9 +40,9 @@ jobs:
         with:
           go-version-file: go.mod
           check-latest: true
-      - uses: golangci/golangci-lint-action@v6
+      - uses: golangci/golangci-lint-action@v7.0.0
         with:
-          version: v1.63.4
+          version: v2.1.2
   build-image:
     name: Build image
     needs: [build, test]

.golangci.yml

@@ -1,28 +1,45 @@
-run:
-  modules-download-mode: readonly
-  allow-parallel-runners: true
-  timeout: 5m
-
-linters:
-  disable-all: true
-  enable:
-    # default linters
-    - errcheck
-    - gosimple
-    - govet
-    - ineffassign
-    - staticcheck
-    - typecheck
-    - unused
-
-    # additional linters
-    - errorlint
-    - errname
-    - gocyclo
-    - goimports
-    - misspell
-    - gofmt
-    - importas
-    - goconst
-    - gocritic
-    - misspell
+version: "2"
+run:
+  modules-download-mode: readonly
+  allow-parallel-runners: true
+linters:
+  default: none
+  enable:
+    - copyloopvar
+    - errcheck
+    - errname
+    - errorlint
+    - goconst
+    - gocritic
+    - gocyclo
+    - govet
+    - importas
+    - ineffassign
+    - misspell
+    - prealloc
+    - staticcheck
+    - unused
+    - usestdlibvars
+    - wastedassign
+    - whitespace
+  exclusions:
+    generated: lax
+    presets:
+      - comments
+      - common-false-positives
+      - legacy
+      - std-error-handling
+    paths:
+      - third_party$
+      - builtin$
+      - examples$
+formatters:
+  enable:
+    - gofmt
+    - goimports
+  exclusions:
+    generated: lax
+    paths:
+      - third_party$
+      - builtin$
+      - examples$

api/v1alpha1/pipelinegate_types.go

@@ -98,7 +98,7 @@ func (pgs *PipelineGateStatus) IsClosed() bool {
 }
 
 func (pgs *PipelineGateStatus) HasJobRef() bool {
-	return !(pgs.JobRef == nil || *pgs.JobRef == console.NamespacedName{})
+	return pgs.JobRef != nil && *pgs.JobRef != console.NamespacedName{}
 }
 
 func (pgs *PipelineGateStatus) GetConsoleGateState() *console.GateState {

cmd/agent/args/args.go

@@ -19,7 +19,12 @@ import (
 )
 
 const (
-	EnvDeployToken = "DEPLOY_TOKEN"
+	EnvDeployToken          = "DEPLOY_TOKEN"
+	EnvDatadogEnabled       = "DATADOG_ENABLED"
+	EnvPyroscopeEnabled     = "PYROSCOPE_ENABLED"
+	EnvProfilerEnabled      = "PROFILER_ENABLED"
+	EnvResourceCacheEnabled = "RESOURCE_CACHE_ENABLED"
+	EnvLocal                = "LOCAL"
 
 	defaultProbeAddress   = ":9001"
 	defaultMetricsAddress = ":8000"
@@ -49,15 +54,21 @@ const (
 
 	defaultProfilerPath    = "/debug/pprof/"
 	defaultProfilerAddress = ":7777"
+
+	defaultPyroscopeAddress = "http://pyroscope.monitoring.svc.cluster.local:4040"
+	defaultDatadogHost      = "datadog-agent.monitoring.svc.cluster.local"
+	defaultDatadogEnv       = "plrl-dev-aws"
 )
 
 var (
 	argDisableHelmTemplateDryRunServer = flag.Bool("disable-helm-dry-run-server", false, "Disable helm template in dry-run=server mode.")
 	argEnableHelmDependencyUpdate      = flag.Bool("enable-helm-dependency-update", false, "Enable update Helm chart's dependencies.")
 	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'")
-	argDisableResourceCache            = flag.Bool("disable-resource-cache", false, "Control whether resource cache should be enabled or not.")
+	argLocal                           = flag.Bool("local", helpers.GetPluralEnvBool(EnvLocal, false), "Whether you're running the operator locally.")
+	argProfiler                        = flag.Bool("profiler", helpers.GetPluralEnvBool(EnvProfilerEnabled, false), "Enable pprof handler. By default it will be exposed on localhost:7777 under '/debug/pprof'")
+	argPyroscope                       = flag.Bool("pyroscope", helpers.GetPluralEnvBool(EnvPyroscopeEnabled, false), "Enable pyroscope integration for detailed application profiling. By default it will push to http://pyroscope.monitoring.svc.cluster.local:4040")
+	argDatadog                         = flag.Bool("datadog", helpers.GetPluralEnvBool(EnvDatadogEnabled, false), "Enable datadog integration for detailed application profiling. By default it will push to http://datadog.monitoring.svc.cluster.local:8125")
+	argDisableResourceCache            = flag.Bool("disable-resource-cache", !helpers.GetPluralEnvBool(EnvResourceCacheEnabled, true), "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.")
 
 	argMaxConcurrentReconciles = flag.Int("max-concurrent-reconciles", 20, "Maximum number of concurrent reconciles which can be run.")
@@ -78,6 +89,9 @@ 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.")
+	argDatadogHost        = flag.String("datadog-host", defaultDatadogHost, "The address of the Datadog server.")
+	argDatadogEnv         = flag.String("datadog-env", defaultDatadogEnv, "The environment of the Datadog server.")
 
 	serviceSet containers.Set[string]
 )
@@ -278,6 +292,26 @@ func ResourceCacheEnabled() bool {
 	return !(*argDisableResourceCache)
 }
 
+func PyroscopeEnabled() bool {
+	return *argPyroscope
+}
+
+func PyroscopeAddress() string {
+	return *argPyroscopeAddress
+}
+
+func DatadogEnabled() bool {
+	return *argDatadog
+}
+
+func DatadogHost() string {
+	return *argDatadogHost
+}
+
+func DatadogEnv() string {
+	return *argDatadogEnv
+}
+
 func ensureOrDie(argName string, arg *string) {
 	if arg == nil || len(*arg) == 0 {
 		pflag.PrintDefaults()

cmd/agent/args/datadog.go

@@ -0,0 +1,48 @@
+package args
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/DataDog/dd-trace-go/v2/ddtrace/tracer"
+	"github.com/DataDog/dd-trace-go/v2/profiler"
+	"k8s.io/klog/v2"
+)
+
+func InitDatadog() error {
+	klog.Info("initializing datadog")
+
+	env := DatadogEnv()
+	service := "deployment-operator"
+	agentAddr := fmt.Sprintf("%s:%s", DatadogHost(), "8126")
+	dogstatsdAddr := fmt.Sprintf("%s:%s", DatadogHost(), "8125")
+
+	if err := tracer.Start(
+		tracer.WithRuntimeMetrics(),
+		tracer.WithDogstatsdAddr(dogstatsdAddr),
+		tracer.WithAgentAddr(agentAddr),
+		tracer.WithService(service),
+		tracer.WithEnv(env),
+	); err != nil {
+		return err
+	}
+
+	return profiler.Start(
+		profiler.WithService(service),
+		profiler.WithEnv(env),
+		profiler.WithTags(fmt.Sprintf("cluster_id:%s", ClusterId()), fmt.Sprintf("console_url:%s", ConsoleUrl())),
+		profiler.WithAgentAddr(agentAddr),
+		profiler.WithPeriod(30*time.Second),
+		profiler.CPUDuration(30*time.Second),
+		profiler.WithProfileTypes(
+			profiler.CPUProfile,
+			profiler.HeapProfile,
+			// The profiles below are disabled by default to keep overhead
+			// low, but can be enabled as needed.
+
+			// profiler.BlockProfile,
+			profiler.MutexProfile,
+			profiler.GoroutineProfile,
+		),
+	)
+}

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/console.go

@@ -4,18 +4,18 @@ import (
 	"os"
 	"time"
 
+	"k8s.io/apimachinery/pkg/runtime"
+
 	"github.com/pluralsh/deployment-operator/cmd/agent/args"
 	"github.com/pluralsh/deployment-operator/internal/utils"
 	"github.com/pluralsh/deployment-operator/pkg/client"
 	consolectrl "github.com/pluralsh/deployment-operator/pkg/controller"
 	"github.com/pluralsh/deployment-operator/pkg/controller/stacks"
 	v1 "github.com/pluralsh/deployment-operator/pkg/controller/v1"
-	"k8s.io/apimachinery/pkg/runtime"
 
 	"k8s.io/client-go/rest"
 	ctrclient "sigs.k8s.io/controller-runtime/pkg/client"
 
-	"github.com/pluralsh/deployment-operator/pkg/controller"
 	"github.com/pluralsh/deployment-operator/pkg/controller/namespaces"
 	"github.com/pluralsh/deployment-operator/pkg/controller/pipelinegates"
 	"github.com/pluralsh/deployment-operator/pkg/controller/restore"
@@ -47,7 +47,7 @@ const (
 )
 
 func registerConsoleReconcilersOrDie(
-	mgr *controller.Manager,
+	mgr *consolectrl.Manager,
 	config *rest.Config,
 	k8sClient ctrclient.Client,
 	scheme *runtime.Scheme,

cmd/agent/kubernetes.go

@@ -12,12 +12,6 @@ import (
 	rolloutv1alpha1 "github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1"
 	roclientset "github.com/argoproj/argo-rollouts/pkg/client/clientset/versioned"
 	cmap "github.com/orcaman/concurrent-map/v2"
-	"github.com/pluralsh/deployment-operator/cmd/agent/args"
-	"github.com/pluralsh/deployment-operator/internal/controller"
-	"github.com/pluralsh/deployment-operator/pkg/cache"
-	consoleclient "github.com/pluralsh/deployment-operator/pkg/client"
-	consolectrl "github.com/pluralsh/deployment-operator/pkg/controller"
-	"github.com/pluralsh/deployment-operator/pkg/controller/service"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	velerov1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -31,6 +25,13 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/metrics/server"
+
+	"github.com/pluralsh/deployment-operator/cmd/agent/args"
+	"github.com/pluralsh/deployment-operator/internal/controller"
+	"github.com/pluralsh/deployment-operator/pkg/cache"
+	consoleclient "github.com/pluralsh/deployment-operator/pkg/client"
+	consolectrl "github.com/pluralsh/deployment-operator/pkg/controller"
+	"github.com/pluralsh/deployment-operator/pkg/controller/service"
 )
 
 const serviceIDCacheExpiry = 12 * time.Hour
@@ -108,7 +109,6 @@ func registerKubeReconcilersOrDie(
 	discoveryClient discovery.DiscoveryInterface,
 	enableKubecostProxy bool,
 ) {
-
 	rolloutsClient, dynamicClient, kubeClient, metricsClient := initKubeClientsOrDie(config)
 
 	backupController := &controller.BackupReconciler{

cmd/agent/main.go

@@ -5,6 +5,9 @@ import (
 	"os"
 	"time"
 
+	kubernetestrace "github.com/DataDog/dd-trace-go/contrib/k8s.io/client-go/v2/kubernetes"
+	datadogtracer "github.com/DataDog/dd-trace-go/v2/ddtrace/tracer"
+	datadogprofiler "github.com/DataDog/dd-trace-go/v2/profiler"
 	trivy "github.com/aquasecurity/trivy-operator/pkg/apis/aquasecurity/v1alpha1"
 	rolloutv1alpha1 "github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1"
 	certmanagerv1 "github.com/cert-manager/cert-manager/pkg/apis/certmanager/v1"
@@ -55,6 +58,33 @@ func main() {
 	config := ctrl.GetConfigOrDie()
 	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()
+		}()
+	}
+
+	if args.DatadogEnabled() {
+		err := args.InitDatadog()
+		if err != nil {
+			panic("unable to initialize datadog")
+		}
+
+		// Trace kubernetes client calls
+		config.WrapTransport = kubernetestrace.WrapRoundTripper
+
+		defer func() {
+			datadogtracer.Stop()
+			datadogprofiler.Stop()
+		}()
+	}
+
 	extConsoleClient := client.New(args.ConsoleUrl(), args.DeployToken())
 	discoveryClient := initDiscoveryClientOrDie(config)
 	kubeManager := initKubeManagerOrDie(config)