[AGENTONB-2759] Use k8s informer for helm metadata (#2488)

* informer on helm CMs/secrets

* add watch processing logic to helm metadata

* remove chartname from label filter

* use sync.Map and add release snapshot ticker

* cleanup unused args and methods

* use workqueue for informer events

* clean up uninstall deletion logic

* reduce logging

* update tests

* fix make fmt

* change ticker to 5 min

* allow helm metadata collection for DAWO

* change chart name

* use manager.Runnable pattern for helm metadata

* use getClient instead of getAPIReader

* fix deletion logic w/ revision number and delete prefix

* use releaseentry with mutex to prevent race condition

* use multiple workers with crash handling

* use 30 sec context timeout for all k8sclient calls

* Update pkg/controller/utils/metadata/helm_metadata.go

Co-authored-by: levan-m <116471169+levan-m@users.noreply.github.com>

* fix syntax issue from merge conflict

* make fmt

---------

Co-authored-by: Levan Machablishvili <levan.machablishvili@datadoghq.com>
Co-authored-by: levan-m <116471169+levan-m@users.noreply.github.com>

Author: swang392

cmd/main.go

@@ -388,13 +388,18 @@ func run(opts *options) error {
 		return setupErrorf(setupLog, err, "Unable to start controllers")
 	}
 
+	// Register Helm metadata forwarder as a manager Runnable
+	// This ensures it starts after cache sync and respects leader election
+	if err = setupAndStartHelmMetadataForwarder(metadataLog, mgr, mgr.GetClient(), versionInfo.String(), options.CredsManager); err != nil {
+		return setupErrorf(setupLog, err, "Unable to setup Helm metadata forwarder")
+	}
+
+	// Start ticker-based metadata forwarders after leader election
 	go func() {
-		// Block until this controller manager is elected leader
 		<-mgr.Elected()
 		setupLog.Info("Starting metadata forwarders")
-		setupAndStartOperatorMetadataForwarder(metadataLog, mgr.GetAPIReader(), versionInfo.String(), opts, options.CredsManager)
-		setupAndStartHelmMetadataForwarder(metadataLog, mgr.GetAPIReader(), versionInfo.String(), options.CredsManager)
-		setupAndStartCRDMetadataForwarder(metadataLog, mgr.GetAPIReader(), versionInfo.String(), opts, options.CredsManager)
+		setupAndStartOperatorMetadataForwarder(metadataLog, mgr.GetClient(), versionInfo.String(), opts, options.CredsManager)
+		setupAndStartCRDMetadataForwarder(metadataLog, mgr.GetClient(), versionInfo.String(), opts, options.CredsManager)
 	}()
 
 	// +kubebuilder:scaffold:builder
@@ -629,7 +634,8 @@ func setupAndStartCRDMetadataForwarder(logger logr.Logger, client client.Reader,
 	cmf.Start()
 }
 
-func setupAndStartHelmMetadataForwarder(logger logr.Logger, client client.Reader, kubernetesVersion string, credsManager *config.CredentialManager) {
-	hmf := metadata.NewHelmMetadataForwarder(logger, client, kubernetesVersion, version.GetVersion(), credsManager)
-	hmf.Start()
+func setupAndStartHelmMetadataForwarder(logger logr.Logger, mgr manager.Manager, client client.Reader, kubernetesVersion string, credsManager *config.CredentialManager) error {
+	hmf := metadata.NewHelmMetadataForwarderWithManager(logger, mgr, client, kubernetesVersion, version.GetVersion(), credsManager)
+	// Register as a runnable with the manager - will be started after cache sync
+	return mgr.Add(hmf)
 }

pkg/controller/utils/metadata/crd_metadata.go

@@ -110,7 +110,10 @@ func (cmf *CRDMetadataForwarder) Start() {
 }
 
 func (cmf *CRDMetadataForwarder) sendMetadata() error {
-	allCRDs, listSuccess := cmf.getAllActiveCRDs()
+	ctx, cancel := context.WithTimeout(context.Background(), DefaultOperationTimeout)
+	defer cancel()
+
+	allCRDs, listSuccess := cmf.getAllActiveCRDs(ctx)
 	crdsToSend := cmf.getCRDsToSend(allCRDs)
 
 	if len(crdsToSend) == 0 {
@@ -122,7 +125,7 @@ func (cmf *CRDMetadataForwarder) sendMetadata() error {
 
 	// Send individual payloads for each CRD
 	for _, crd := range crdsToSend {
-		if err := cmf.sendCRDMetadata(crd); err != nil {
+		if err := cmf.sendCRDMetadata(ctx, crd); err != nil {
 			cmf.logger.V(1).Info("Failed to send metadata", "error", err,
 				"kind", crd.Kind, "name", crd.Name, "namespace", crd.Namespace)
 		}
@@ -132,8 +135,8 @@ func (cmf *CRDMetadataForwarder) sendMetadata() error {
 	return nil
 }
 
-func (cmf *CRDMetadataForwarder) sendCRDMetadata(crdInstance CRDInstance) error {
-	clusterUID, err := cmf.GetOrCreateClusterUID()
+func (cmf *CRDMetadataForwarder) sendCRDMetadata(ctx context.Context, crdInstance CRDInstance) error {
+	clusterUID, err := cmf.GetOrCreateClusterUID(ctx)
 	if err != nil {
 		return fmt.Errorf("error getting cluster UID: %w", err)
 	}
@@ -149,6 +152,8 @@ func (cmf *CRDMetadataForwarder) sendCRDMetadata(crdInstance CRDInstance) error
 		return fmt.Errorf("error creating request: %w", err)
 	}
 
+	req = req.WithContext(ctx)
+
 	resp, err := cmf.httpClient.Do(req)
 	if err != nil {
 		return fmt.Errorf("error sending metadata request: %w", err)
@@ -224,7 +229,7 @@ func (cmf *CRDMetadataForwarder) buildPayload(clusterUID string, crdInstance CRD
 
 // getAllActiveCRDs returns all active CRDs and a map of list successes for each CRD type
 // Currently only DatadogAgent, DatadogAgentInternal, and DatadogAgentProfile are collected
-func (cmf *CRDMetadataForwarder) getAllActiveCRDs() ([]CRDInstance, map[string]bool) {
+func (cmf *CRDMetadataForwarder) getAllActiveCRDs(ctx context.Context) ([]CRDInstance, map[string]bool) {
 	var crds []CRDInstance
 	listSuccess := make(map[string]bool)
 
@@ -234,7 +239,7 @@ func (cmf *CRDMetadataForwarder) getAllActiveCRDs() ([]CRDInstance, map[string]b
 	// DDA
 	if cmf.enabledCRDs.DatadogAgentEnabled {
 		ddaList := &v2alpha1.DatadogAgentList{}
-		if err := cmf.k8sClient.List(context.TODO(), ddaList); err == nil {
+		if err := cmf.k8sClient.List(ctx, ddaList); err == nil {
 			listSuccess["DatadogAgent"] = true
 			for _, dda := range ddaList.Items {
 				annotations := maps.Clone(dda.Annotations)
@@ -259,7 +264,7 @@ func (cmf *CRDMetadataForwarder) getAllActiveCRDs() ([]CRDInstance, map[string]b
 	// DDAI
 	if cmf.enabledCRDs.DatadogAgentInternalEnabled {
 		ddaiList := &v1alpha1.DatadogAgentInternalList{}
-		if err := cmf.k8sClient.List(context.TODO(), ddaiList); err == nil {
+		if err := cmf.k8sClient.List(ctx, ddaiList); err == nil {
 			listSuccess["DatadogAgentInternal"] = true
 			for _, ddai := range ddaiList.Items {
 				annotations := maps.Clone(ddai.Annotations)
@@ -284,7 +289,7 @@ func (cmf *CRDMetadataForwarder) getAllActiveCRDs() ([]CRDInstance, map[string]b
 	// DAP
 	if cmf.enabledCRDs.DatadogAgentProfileEnabled {
 		dapList := &v1alpha1.DatadogAgentProfileList{}
-		if err := cmf.k8sClient.List(context.TODO(), dapList); err == nil {
+		if err := cmf.k8sClient.List(ctx, dapList); err == nil {
 			listSuccess["DatadogAgentProfile"] = true
 			for _, dap := range dapList.Items {
 				annotations := maps.Clone(dap.Annotations)

pkg/controller/utils/metadata/credential_setup_test.go

@@ -6,6 +6,7 @@
 package metadata
 
 import (
+	"context"
 	"os"
 	"strings"
 	"testing"
@@ -334,7 +335,7 @@ func TestSetupRequestPrerequisites(t *testing.T) {
 			assert.Equal(t, "application/json", req.Header.Get("Accept"), "Accept header should be set")
 
 			// Verify cluster UID is set
-			clusterUID, err := omf.GetOrCreateClusterUID()
+			clusterUID, err := omf.GetOrCreateClusterUID(context.Background())
 			assert.NoError(t, err)
 			assert.NotEmpty(t, clusterUID, "Cluster UID should be set")
 		})