feat(cluster-agent): enhance DatadogPodAutoscaler metrics with leader checks and tests

  Improve the metrics generation and submission for DatadogPodAutoscaler:

  * **Observer pattern enhancement**: Changed autoscaling.ObserverFunc signature
    from func(string, string) to func(string, interface{}) to allow observers
    direct access to stored objects without additional lookups

  * **Centralized tag generation**: Refactored generator.go to use helper
    functions (baseAutoscalerTags, autoscalerTagsWithSource,
    autoscalerTagsWithContainer, conditionTags) making it easier to add or
    modify tags across all metrics

  * **Leader-only metric submission**: Added isLeader checks to
    SenderMetricsWriter, horizontalController, and verticalController to
    prevent non-leader instances from submitting duplicate metrics

  * **Backward compatibility**: Retained isLeader:true tag on all metrics
    via baseAutoscalerTags helper function

Author: clamoriniere

.github/CODEOWNERS

@@ -464,6 +464,7 @@
 /pkg/cloudfoundry                       @DataDog/agent-integrations
 /pkg/clusteragent/                      @DataDog/container-platform
 /pkg/clusteragent/autoscaling/          @DataDog/container-autoscaling
+/pkg/clusteragent/metricsstore/         @DataDog/container-autoscaling
 /pkg/clusteragent/admission/mutate/autoscaling          @DataDog/container-integrations
 /pkg/clusteragent/admission/mutate/appsec               @DataDog/container-platform @DataDog/asm-go
 /pkg/clusteragent/admission/mutate/autoinstrumentation/ @DataDog/container-platform @DataDog/injection-platform

cmd/cluster-agent/subcommands/start/command.go

@@ -538,7 +538,7 @@ func start(log log.Component,
 			log.Error("Admission controller is disabled, vertical autoscaling requires the admission controller to be enabled. Vertical scaling will be disabled.")
 		}
 
-		if adapter, err := provider.StartWorkloadAutoscaling(mainCtx, clusterID, clusterName, le.IsLeader, apiCl, rcClient, wmeta, demultiplexer); err == nil {
+		if adapter, err := provider.StartWorkloadAutoscaling(mainCtx, clusterID, clusterName, le.IsLeader, apiCl, rcClient, wmeta, taggerComp, demultiplexer); err == nil {
 			pa = adapter
 		} else {
 			return fmt.Errorf("Error while starting workload autoscaling: %v", err)

pkg/clusteragent/autoscaling/cluster/config_retriever.go

@@ -12,13 +12,14 @@ import (
 
 	"k8s.io/utils/clock"
 
+	"github.com/DataDog/datadog-agent/pkg/clusteragent/autoscaling"
 	"github.com/DataDog/datadog-agent/pkg/config/remote/data"
 	"github.com/DataDog/datadog-agent/pkg/remoteconfig/state"
 	"github.com/DataDog/datadog-agent/pkg/util/log"
 )
 
 const (
-	configRetrieverStoreID string = "cacr"
+	configRetrieverStoreID autoscaling.SenderID = "cacr"
 )
 
 // RcClient is a subinterface of rcclient.Component to allow mocking

pkg/clusteragent/autoscaling/cluster/controller.go

@@ -40,7 +40,7 @@ var (
 	// Only support EC2 for now
 	nodeClassGVR = schema.GroupVersionResource{Group: "karpenter.k8s.aws", Version: "v1", Resource: "ec2nodeclasses"}
 
-	controllerID = "dca-c"
+	controllerID autoscaling.SenderID = "dca-c"
 )
 
 type Controller struct {

pkg/clusteragent/autoscaling/cluster/telemetry.go

@@ -16,5 +16,3 @@ const (
 )
 
 var autoscalingQueueMetricsProvider = workqueuetelemetry.NewQueueMetricsProvider()
-
-// TODO add more telemetry collection

pkg/clusteragent/autoscaling/controller.go

@@ -30,7 +30,7 @@ type Controller struct {
 	context   context.Context
 
 	// Fields available to child controllers
-	ID        string
+	ID        SenderID
 	Client    dynamic.Interface
 	Lister    cache.GenericLister
 	Workqueue workqueue.TypedRateLimitingInterface[string]
@@ -39,7 +39,7 @@ type Controller struct {
 
 // NewController returns a new workload autoscaling controller
 func NewController(
-	controllerID string,
+	controllerID SenderID,
 	processor Processor,
 	client dynamic.Interface,
 	informer dynamicinformer.DynamicSharedInformerFactory,
@@ -120,7 +120,7 @@ func (c *Controller) enqueue(obj interface{}) {
 	c.Workqueue.AddRateLimited(key)
 }
 
-func (c *Controller) enqueueID(id, sender string) {
+func (c *Controller) enqueueID(id string, sender SenderID, _ interface{}) {
 	// Do not enqueue our own updates (avoid infinite loops)
 	if sender != c.ID {
 		log.Tracef("Enqueueing from observer update id: %s from sender: %s", id, sender)

pkg/clusteragent/autoscaling/externalmetrics/autoscaler_watcher.go

@@ -27,18 +27,19 @@ import (
 
 	"github.com/DataDog/watermarkpodautoscaler/apis/datadoghq/v1alpha1"
 
+	"github.com/DataDog/datadog-agent/pkg/clusteragent/autoscaling"
 	"github.com/DataDog/datadog-agent/pkg/clusteragent/autoscaling/externalmetrics/model"
 	"github.com/DataDog/datadog-agent/pkg/util/kubernetes/apiserver/controllers"
 	"github.com/DataDog/datadog-agent/pkg/util/kubernetes/autoscalers"
 	"github.com/DataDog/datadog-agent/pkg/util/log"
 )
 
 const (
-	autoscalerWatcherStoreID    string = "aw"
-	autoscalerReferencesSep     string = ", "
-	autoscalerReferencesKindSep string = ":"
-	autoscalerWPAKindKey        string = "wpa"
-	autoscalerHPAKindKey        string = "hpa"
+	autoscalerWatcherStoreID    autoscaling.SenderID = "aw"
+	autoscalerReferencesSep     string               = ", "
+	autoscalerReferencesKindSep string               = ":"
+	autoscalerWPAKindKey        string               = "wpa"
+	autoscalerHPAKindKey        string               = "hpa"
 )
 
 // AutoscalerWatcher watches autoscaling objects and reconciles the corresponding external metrics

pkg/clusteragent/autoscaling/externalmetrics/datadogmetric_controller.go

@@ -13,6 +13,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/DataDog/datadog-agent/pkg/clusteragent/autoscaling"
 	"github.com/DataDog/datadog-agent/pkg/clusteragent/autoscaling/externalmetrics/model"
 	le "github.com/DataDog/datadog-agent/pkg/util/kubernetes/apiserver/leaderelection/metrics"
 	"github.com/DataDog/datadog-agent/pkg/util/log"
@@ -29,9 +30,9 @@ import (
 )
 
 const (
-	maxRetry             int    = 3
-	requeueDelaySeconds  int    = 2
-	ddmControllerStoreID string = "ddmc"
+	maxRetry             int                  = 3
+	requeueDelaySeconds  int                  = 2
+	ddmControllerStoreID autoscaling.SenderID = "ddmc"
 )
 
 type controllerOperation string
@@ -146,8 +147,8 @@ func (c *DatadogMetricController) enqueue(obj interface{}) {
 	c.workqueue.AddRateLimited(key)
 }
 
-func (c *DatadogMetricController) enqueueID(id, sender string) {
-	// Do not enqueue our own updates (avoid infinite loops)
+func (c *DatadogMetricController) enqueueID(id string, sender autoscaling.SenderID, _ interface{}) {
+	// Avoid re-enqueuing events triggered by this controller's own store updates
 	if sender != ddmControllerStoreID {
 		c.workqueue.AddRateLimited(id)
 	}
@@ -352,7 +353,7 @@ func (c *DatadogMetricController) deleteDatadogMetric(ns, name string) error {
 	return nil
 }
 
-func (c *DatadogMetricController) deleteTelemetry(id, _ string) {
+func (c *DatadogMetricController) deleteTelemetry(id string, _ autoscaling.SenderID, _ interface{}) {
 	ns, name, err := cache.SplitMetaNamespaceKey(id)
 	if err != nil {
 		log.Debugf("Unable to split meta namespace key to delete telemetry: %v", err)

pkg/clusteragent/autoscaling/externalmetrics/metrics_retriever.go

@@ -11,6 +11,7 @@ import (
 	"errors"
 	"time"
 
+	"github.com/DataDog/datadog-agent/pkg/clusteragent/autoscaling"
 	"github.com/DataDog/datadog-agent/pkg/clusteragent/autoscaling/externalmetrics/model"
 	"github.com/DataDog/datadog-agent/pkg/util/backoff"
 	le "github.com/DataDog/datadog-agent/pkg/util/kubernetes/apiserver/leaderelection/metrics"
@@ -19,7 +20,7 @@ import (
 )
 
 const (
-	metricRetrieverStoreID string = "mr"
+	metricRetrieverStoreID autoscaling.SenderID = "mr"
 )
 
 // Backoff range for number of retries R:

pkg/clusteragent/autoscaling/externalmetrics/store.go

@@ -10,6 +10,7 @@ package externalmetrics
 import (
 	"sync"
 
+	"github.com/DataDog/datadog-agent/pkg/clusteragent/autoscaling"
 	"github.com/DataDog/datadog-agent/pkg/clusteragent/autoscaling/externalmetrics/model"
 )
 
@@ -19,7 +20,8 @@ const (
 )
 
 // DatadogMetricInternalObserverFunc represents observer functions of the datadog metrics store
-type DatadogMetricInternalObserverFunc func(string, string)
+// First parameter is the key, second parameter is the sender identifier, third parameter is the object (or nil for delete operations)
+type DatadogMetricInternalObserverFunc func(string, autoscaling.SenderID, interface{})
 
 // DatadogMetricInternalObserver allows to define functions to watch changes in Store
 type DatadogMetricInternalObserver struct {
@@ -108,23 +110,23 @@ func (ds *DatadogMetricsInternalStore) Count() int {
 }
 
 // Set `DatadogMetricInternal` for id
-func (ds *DatadogMetricsInternalStore) Set(id string, datadogMetric model.DatadogMetricInternal, sender string) {
+func (ds *DatadogMetricsInternalStore) Set(id string, datadogMetric model.DatadogMetricInternal, sender autoscaling.SenderID) {
 	ds.lock.Lock()
 	ds.store[id] = datadogMetric
 	ds.lock.Unlock()
 
-	ds.notify(setOperation, id, sender)
+	ds.notify(setOperation, id, sender, datadogMetric)
 }
 
 // Delete `DatadogMetricInternal` corresponding to id if present
-func (ds *DatadogMetricsInternalStore) Delete(id, sender string) {
+func (ds *DatadogMetricsInternalStore) Delete(id string, sender autoscaling.SenderID) {
 	ds.lock.Lock()
-	_, exists := ds.store[id]
+	obj, exists := ds.store[id]
 	delete(ds.store, id)
 	ds.lock.Unlock()
 
 	if exists {
-		ds.notify(deleteOperation, id, sender)
+		ds.notify(deleteOperation, id, sender, obj)
 	}
 }
 
@@ -152,31 +154,31 @@ func (ds *DatadogMetricsInternalStore) Unlock(string) {
 }
 
 // UnlockSet sets the new DatadogMetricInternal value and releases the lock (previously acquired by `LockRead`)
-func (ds *DatadogMetricsInternalStore) UnlockSet(id string, datadogMetric model.DatadogMetricInternal, sender string) {
+func (ds *DatadogMetricsInternalStore) UnlockSet(id string, datadogMetric model.DatadogMetricInternal, sender autoscaling.SenderID) {
 	ds.store[id] = datadogMetric
 	ds.lock.Unlock()
 
-	ds.notify(setOperation, id, sender)
+	ds.notify(setOperation, id, sender, datadogMetric)
 }
 
 // UnlockDelete deletes a DatadogMetricInternal and releases the lock (previously acquired by `LockRead`)
-func (ds *DatadogMetricsInternalStore) UnlockDelete(id, sender string) {
-	_, exists := ds.store[id]
+func (ds *DatadogMetricsInternalStore) UnlockDelete(id string, sender autoscaling.SenderID) {
+	obj, exists := ds.store[id]
 
 	delete(ds.store, id)
 	ds.lock.Unlock()
 
 	if exists {
-		ds.notify(deleteOperation, id, sender)
+		ds.notify(deleteOperation, id, sender, obj)
 	}
 }
 
 // It's a very simple implementation of a notify process, but it's enough in our case as we aim at only 1 or 2 observers
-func (ds *DatadogMetricsInternalStore) notify(operationType storeOperation, key, sender string) {
+func (ds *DatadogMetricsInternalStore) notify(operationType storeOperation, key string, sender autoscaling.SenderID, obj interface{}) {
 	ds.observersLock.RLock()
 	defer ds.observersLock.RUnlock()
 
 	for _, observer := range ds.observers[operationType] {
-		observer(key, sender)
+		observer(key, sender, obj)
 	}
 }