[CASCL-623] feat(cluster-agent): enhance DatadogPodAutoscaler metrics with dedicate metrics store (#46833)

## Scope of This PR

This PR revisits the previous attempt:  
#42547

The focus here is strictly on **refactoring the existing metric generation logic**, without introducing new metrics.

A follow-up PR will build on this foundation to introduce additional DPA metrics.

## Motivation

Today, `DatadogPodAutoscaler` (DPA) resource metrics are exposed as OpenMetrics/Prometheus metrics and scraped by the `datadog_cluster_agent` check.

While functional, this approach has several drawbacks:

- All scraped metrics automatically inherit the **Cluster Agent pod context** (`pod_name`, `kube_namespace`, `kube_container_name`, etc.).
- This creates confusion around DPA metric tags, as metrics appear tied to Cluster Agent pods rather than the DPA resources themselves.
- It increases metric cardinality due to multiple Cluster Agent instances (leader, followers, rollouts) contributing additional metric contexts.

## Proposed Approach

This PR changes how DPA metrics are generated.

Instead of exposing them via OpenMetrics and relying on the `datadog_cluster_agent` check for collection, DPA metrics are now produced directly within the Cluster Agent autoscaling component — following the same pattern used for `kubernetes_state` metrics.

This provides:
- Better control over which tags are attached to DPA metrics  
- Cleaner and more accurate metric context  
- Reduced unnecessary metric cardinality  

## Summary

### Core changes
- Refactor `ObserverFunc` signature from `(string, string)` to `(string, interface{})` to pass the actual object to observers, enabling richer metric generation
- Add new `pkg/clusteragent/autoscaling/workload/metrics` package with a `PodAutoscalerMetricsStore` that generates and periodically sends structured metrics (gauges/counts) for `DatadogPodAutoscaler` objects via `sender.Sender`
- Replace old telemetry helpers (`telemetry.go` tag-based metrics) with leader-aware metric submission; metrics are only emitted by the leader

### Action metrics consolidation
- Move horizontal/vertical scaling action metrics from event-driven `Submit*` functions in `counters.go` into the state-based `GeneratePodAutoscalerMetrics` generator
- Delete `counters.go` entirely (`SubmitReceivedRecommendationsVersion` was dead code; remaining `Submit*` functions replaced by the generator)
- Remove `sender` and `isLeader` from `horizontalController` and `verticalController` since they were only used for the now-removed `Submit*` calls

### Metrics emitted
| Metric | Type | Notes |
|--------|------|-------|
| `received_recommendations_version` | Gauge | RC version of last received main scaling values; only emitted when > 0 |
| `horizontal_scaling_received_replicas` | Gauge | Replicas recommended by the product recommender |
| `vertical_scaling_received_requests` | Gauge | Per-container requested resources from recommender |
| `vertical_scaling_received_limits` | Gauge | Per-container resource limits from recommender |
| `horizontal_scaling_applied_replicas` | Gauge | Replicas from the last applied horizontal action |
| `horizontal_scaling_actions` | Count | Cumulative count of horizontal scaling actions, tagged `status:ok` or `status:error` |
| `vertical_rollout_triggered` | Count | Cumulative count of vertical rollout actions, tagged `status:ok` or `status:error` |
| `autoscaler_conditions` | Gauge | 1.0/0.0 per condition type from CRD status |
| `local_fallback_enabled` | Gauge | 1.0 when horizontal active source is local fallback |

### Model changes
- Add `mainScalingValuesVersion uint64` to `PodAutoscalerInternal` to persist the remote config version of the last received main scaling values
- Extend `UpdateFromMainValues` to accept and persist the RC version; `RemoveMainValues` resets it to 0
- Add `horizontalActionErrorCount`/`horizontalActionSuccessCount` and `verticalActionErrorCount`/`verticalActionSuccessCount` counter fields, incremented on each action outcome
- Add public getters: `MainScalingValuesVersion()`, `HorizontalActionErrorCount()`, `HorizontalActionSuccessCount()`, `VerticalActionErrorCount()`, `VerticalActionSuccessCount()`

## Test plan

- [x] Unit tests added for `metrics/store`, `metrics/generator`, and `metrics/writer`
- [x] `generator_test.go` covers all 9 metrics including both `status:ok` and `status:error` count variants
- [x] `config_retriever_values_test.go` updated with expected `MainScalingValuesVersion` values
- [x] `controller_horizontal_test.go` updated to assert `HorizontalActionSuccessCount`/`HorizontalActionErrorCount`
- [x] Existing `workload` controller tests updated to remove now-deleted `sender`/`isLeader` fixtures
- [x] All tests in `pkg/clusteragent/autoscaling/...` pass locally

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: cedric.lamoriniere <cedric.lamoriniere@datadoghq.com>

Author: clamoriniere

.github/CODEOWNERS

@@ -466,6 +466,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)
 	}
 }