[autoscaling] Add error reasons in the autoscaling conditions (#46063)

### What does this PR do?

Add codified `Reason` in the DPA Status to allow better error matching and understanding

### Motivation

### Describe how you validated your changes

There are a significant number of error cases, testing everything is not required.
To test an easy one, you can create an Autoscaler that references an unsupported `apiGroup`

### Additional Notes


Co-authored-by: vincent.boulineau <vincent.boulineau@datadoghq.com>

Author: vboulineau

pkg/clusteragent/autoscaling/errors.go

@@ -0,0 +1,74 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016-present Datadog, Inc.
+
+//go:build kubeapiserver
+
+package autoscaling
+
+import (
+	"fmt"
+)
+
+// ConditionReasonType is a typed string for programmatic condition reasons.
+// Values must be CamelCase with no spaces, following the Kubernetes convention.
+type ConditionReasonType string
+
+const (
+	// ConditionReasonInvalidTargetRef indicates the targetRef API version could not be parsed.
+	ConditionReasonInvalidTargetRef ConditionReasonType = "InvalidTargetRef"
+	// ConditionReasonInvalidTarget indicates the target cannot be autoscaled (e.g. it is the cluster agent itself).
+	ConditionReasonInvalidTarget ConditionReasonType = "InvalidTarget"
+	// ConditionReasonInvalidSpec indicates the DPA spec failed validation (e.g. bad objectives or constraints).
+	ConditionReasonInvalidSpec ConditionReasonType = "InvalidSpec"
+	// ConditionReasonClusterAutoscalerLimitReached indicates the maximum number of DPA objects per cluster has been reached.
+	ConditionReasonClusterAutoscalerLimitReached ConditionReasonType = "ClusterAutoscalerLimitReached"
+	// ConditionReasonRecommendationError indicates Datadog could not compute recommendations.
+	ConditionReasonRecommendationError ConditionReasonType = "RecommendationError"
+	// ConditionReasonLocalRecommenderError indicates the local fallback recommender could not compute recommendations.
+	ConditionReasonLocalRecommenderError ConditionReasonType = "LocalRecommenderError"
+	// ConditionReasonTargetNotFound indicates the scale target could not be found.
+	ConditionReasonTargetNotFound ConditionReasonType = "TargetNotFound"
+	// ConditionReasonScaleFailed indicates the scale operation on the target failed.
+	ConditionReasonScaleFailed ConditionReasonType = "ScaleFailed"
+	// ConditionReasonScalingDisabled indicates scaling is blocked because current replicas is 0.
+	ConditionReasonScalingDisabled ConditionReasonType = "ScalingDisabled"
+	// ConditionReasonPolicyRestricted indicates the applyPolicy or strategy blocks the scaling action.
+	ConditionReasonPolicyRestricted ConditionReasonType = "PolicyRestricted"
+	// ConditionReasonFallbackRestricted indicates the fallback scaling direction is not allowed.
+	ConditionReasonFallbackRestricted ConditionReasonType = "FallbackRestricted"
+	// ConditionReasonUnsupportedTargetKind indicates vertical rollout is not supported for this workload Kind.
+	ConditionReasonUnsupportedTargetKind ConditionReasonType = "UnsupportedTargetKind"
+	// ConditionReasonRolloutFailed indicates a failure when triggering a vertical rollout on the target.
+	ConditionReasonRolloutFailed ConditionReasonType = "RolloutFailed"
+)
+
+// ConditionReason is an interface that errors can implement to provide
+// a programmatic Reason for Kubernetes conditions.
+// When an error implements this interface, Reason() populates the condition's
+// Reason field and Error() populates the Message field.
+// When an error does not implement this interface, Error() populates the
+// Message field and Reason is left empty.
+type ConditionReason interface {
+	Reason() ConditionReasonType
+}
+
+// NewConditionError wraps an existing error with a programmatic reason.
+func NewConditionError(reason ConditionReasonType, err error) error {
+	return &conditionError{reason: reason, err: err}
+}
+
+// NewConditionErrorf creates a new error with a programmatic reason and a formatted message.
+func NewConditionErrorf(reason ConditionReasonType, format string, args ...any) error {
+	return &conditionError{reason: reason, err: fmt.Errorf(format, args...)}
+}
+
+type conditionError struct {
+	reason ConditionReasonType
+	err    error
+}
+
+func (e *conditionError) Error() string               { return e.err.Error() }
+func (e *conditionError) Unwrap() error               { return e.err }
+func (e *conditionError) Reason() ConditionReasonType { return e.reason }

pkg/clusteragent/autoscaling/max_heap.go

@@ -11,12 +11,8 @@ import (
 	"container/heap"
 	"sync"
 	"time"
-
-	"github.com/DataDog/datadog-agent/pkg/clusteragent/autoscaling/workload/model"
 )
 
-type store = Store[model.PodAutoscalerInternal]
-
 // TimestampKey is a struct that holds a timestamp and key for a `DatadogPodAutoscaler` object
 type TimestampKey struct {
 	Timestamp time.Time
@@ -79,22 +75,25 @@ func (h *MaxTimestampKeyHeap) FindIdx(key string) (int, bool) {
 }
 
 // HashHeap is a struct that holds a MaxHeap and a set of keys that exist in the heap
-type HashHeap struct {
-	MaxHeap MaxTimestampKeyHeap
-	Keys    map[string]bool
-	maxSize int
-	mu      sync.RWMutex
-	store   *store
+type HashHeap[T any] struct {
+	MaxHeap           MaxTimestampKeyHeap
+	Keys              map[string]bool
+	maxSize           int
+	mu                sync.RWMutex
+	store             *Store[T]
+	creationTimestamp func(*T) time.Time
 }
 
-// NewHashHeap returns a new MaxHeap with the given max size
-func NewHashHeap(maxSize int, store *store) *HashHeap {
-	h := &HashHeap{
-		MaxHeap: *NewMaxHeap(),
-		Keys:    make(map[string]bool),
-		maxSize: maxSize,
-		mu:      sync.RWMutex{},
-		store:   store,
+// NewHashHeap returns a new MaxHeap with the given max size.
+// creationTimestamp is used to extract the creation timestamp from stored objects.
+func NewHashHeap[T any](maxSize int, store *Store[T], creationTimestamp func(*T) time.Time) *HashHeap[T] {
+	h := &HashHeap[T]{
+		MaxHeap:           *NewMaxHeap(),
+		Keys:              make(map[string]bool),
+		maxSize:           maxSize,
+		mu:                sync.RWMutex{},
+		store:             store,
+		creationTimestamp: creationTimestamp,
 	}
 
 	store.RegisterObserver(Observer{
@@ -107,19 +106,19 @@ func NewHashHeap(maxSize int, store *store) *HashHeap {
 
 // InsertIntoHeap returns true if the key already exists in the max heap or was inserted correctly
 // Used as an ObserverFunc; accept sender as parameter to match ObserverFunc signature
-func (h *HashHeap) InsertIntoHeap(key, _sender string) {
-	// Get object from store
-	podAutoscalerInternal, podAutoscalerInternalFound := h.store.Get(key)
-	if !podAutoscalerInternalFound {
+func (h *HashHeap[T]) InsertIntoHeap(key, _sender string) {
+	obj, found := h.store.Get(key)
+	if !found {
 		return
 	}
 
-	if podAutoscalerInternal.CreationTimestamp().IsZero() {
+	ts := h.creationTimestamp(&obj)
+	if ts.IsZero() {
 		return
 	}
 
 	newTimestampKey := TimestampKey{
-		Timestamp: podAutoscalerInternal.CreationTimestamp(),
+		Timestamp: ts,
 		Key:       key,
 	}
 
@@ -147,7 +146,7 @@ func (h *HashHeap) InsertIntoHeap(key, _sender string) {
 
 // DeleteFromHeap removes the given key from the max heap
 // Used as an ObserverFunc; accept sender as parameter to match ObserverFunc signature
-func (h *HashHeap) DeleteFromHeap(key, _sender string) {
+func (h *HashHeap[T]) DeleteFromHeap(key, _sender string) {
 	// Key did not exist in heap, return early
 	if !h.Exists(key) {
 		return
@@ -167,14 +166,14 @@ func (h *HashHeap) DeleteFromHeap(key, _sender string) {
 }
 
 // Exists returns true if the given key exists in the heap
-func (h *HashHeap) Exists(key string) bool {
+func (h *HashHeap[T]) Exists(key string) bool {
 	h.mu.RLock()
 	defer h.mu.RUnlock()
 	_, ok := h.Keys[key]
 	return ok
 }
 
 // MaxSize returns the size limit on the hash heap
-func (h *HashHeap) MaxSize() int {
+func (h *HashHeap[T]) MaxSize() int {
 	return h.maxSize
 }

pkg/clusteragent/autoscaling/workload/controller.go

@@ -54,7 +54,10 @@ var (
 	}
 )
 
-type store = autoscaling.Store[model.PodAutoscalerInternal]
+type (
+	store     = autoscaling.Store[model.PodAutoscalerInternal]
+	limitHeap = autoscaling.HashHeap[model.PodAutoscalerInternal]
+)
 
 // Controller for DatadogPodAutoscaler objects
 type Controller struct {
@@ -66,7 +69,7 @@ type Controller struct {
 	eventRecorder record.EventRecorder
 	store         *store
 
-	limitHeap *autoscaling.HashHeap
+	limitHeap *limitHeap
 
 	podWatcher           PodWatcher
 	horizontalController *horizontalController
@@ -90,7 +93,7 @@ func NewController(
 	store *store,
 	podWatcher PodWatcher,
 	localSender sender.Sender,
-	limitHeap *autoscaling.HashHeap,
+	limitHeap *limitHeap,
 ) (*Controller, error) {
 	c := &Controller{
 		clusterID:         clusterID,
@@ -461,7 +464,7 @@ func (c *Controller) validateAutoscaler(podAutoscalerInternal model.PodAutoscale
 	// Check that we are within the limit of 100 DatadogPodAutoscalers
 	key := podAutoscalerInternal.ID()
 	if !c.limitHeap.Exists(key) {
-		return fmt.Errorf("Autoscaler disabled as maximum number per cluster reached (%d)", c.limitHeap.MaxSize())
+		return autoscaling.NewConditionErrorf(autoscaling.ConditionReasonClusterAutoscalerLimitReached, "Autoscaler disabled as maximum number per cluster reached (%d)", c.limitHeap.MaxSize())
 	}
 
 	// Check that targetRef is not set to the cluster agent
@@ -485,7 +488,7 @@ func (c *Controller) validateAutoscaler(podAutoscalerInternal model.PodAutoscale
 	clusterAgentNs := namespace.GetMyNamespace()
 
 	if podAutoscalerInternal.Namespace() == clusterAgentNs && podAutoscalerInternal.Spec().TargetRef.Name == resourceName {
-		return errors.New("Autoscaling target cannot be set to the cluster agent")
+		return autoscaling.NewConditionErrorf(autoscaling.ConditionReasonInvalidTarget, "Autoscaling target cannot be set to the cluster agent")
 	}
 	if err := validateAutoscalerObjectives(podAutoscalerInternal.Spec()); err != nil {
 		return err
@@ -497,7 +500,7 @@ func validateAutoscalerObjectives(spec *datadoghq.DatadogPodAutoscalerSpec) erro
 	if spec.Fallback != nil && len(spec.Fallback.Horizontal.Objectives) > 0 {
 		for _, objective := range spec.Fallback.Horizontal.Objectives {
 			if objective.Type == datadoghqcommon.DatadogPodAutoscalerCustomQueryObjectiveType {
-				return errors.New("Autoscaler fallback cannot be based on custom query objective")
+				return autoscaling.NewConditionErrorf(autoscaling.ConditionReasonInvalidSpec, "Autoscaler fallback cannot be based on custom query objective")
 			}
 		}
 	}
@@ -506,15 +509,15 @@ func validateAutoscalerObjectives(spec *datadoghq.DatadogPodAutoscalerSpec) erro
 		switch objective.Type {
 		case datadoghqcommon.DatadogPodAutoscalerCustomQueryObjectiveType:
 			if objective.CustomQuery == nil {
-				return errors.New("Autoscaler objective type is custom query but customQueryObjective is nil")
+				return autoscaling.NewConditionErrorf(autoscaling.ConditionReasonInvalidSpec, "Autoscaler objective type is custom query but customQueryObjective is nil")
 			}
 		case datadoghqcommon.DatadogPodAutoscalerPodResourceObjectiveType:
 			if objective.PodResource == nil {
-				return fmt.Errorf("autoscaler objective type is %s but podResource is nil", objective.Type)
+				return autoscaling.NewConditionErrorf(autoscaling.ConditionReasonInvalidSpec, "autoscaler objective type is %s but podResource is nil", objective.Type)
 			}
 		case datadoghqcommon.DatadogPodAutoscalerContainerResourceObjectiveType:
 			if objective.ContainerResource == nil {
-				return fmt.Errorf("autoscaler objective type is %s but containerResource is nil", objective.Type)
+				return autoscaling.NewConditionErrorf(autoscaling.ConditionReasonInvalidSpec, "autoscaler objective type is %s but containerResource is nil", objective.Type)
 			}
 		}
 	}

pkg/clusteragent/autoscaling/workload/controller_horizontal.go

@@ -9,7 +9,6 @@ package workload
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"math"
 	"slices"
@@ -77,7 +76,7 @@ func (hr *horizontalController) sync(ctx context.Context, podAutoscaler *datadog
 	// Get the current scale of the target resource
 	scale, gr, err := hr.scaler.get(ctx, autoscalerInternal.Namespace(), autoscalerInternal.Spec().TargetRef.Name, gvk)
 	if err != nil {
-		err = fmt.Errorf("failed to get scale subresource for autoscaler %s, err: %w", autoscalerInternal.ID(), err)
+		err = autoscaling.NewConditionError(autoscaling.ConditionReasonTargetNotFound, fmt.Errorf("failed to get scale subresource for autoscaler %s, err: %w", autoscalerInternal.ID(), err))
 		autoscalerInternal.UpdateFromHorizontalAction(nil, err)
 		return autoscaling.Requeue, err
 	}
@@ -127,7 +126,7 @@ func (hr *horizontalController) performScaling(ctx context.Context, podAutoscale
 	scale.Spec.Replicas = horizontalAction.ToReplicas
 	_, err = hr.scaler.update(ctx, gr, scale)
 	if err != nil {
-		err = fmt.Errorf("failed to scale target: %s/%s to %d replicas, err: %w", scale.Namespace, scale.Name, horizontalAction.ToReplicas, err)
+		err = autoscaling.NewConditionError(autoscaling.ConditionReasonScaleFailed, fmt.Errorf("failed to scale target: %s/%s to %d replicas, err: %w", scale.Namespace, scale.Name, horizontalAction.ToReplicas, err))
 		hr.eventRecorder.Event(podAutoscaler, corev1.EventTypeWarning, model.FailedScaleEventReason, err.Error())
 		autoscalerInternal.UpdateFromHorizontalAction(nil, err)
 
@@ -161,7 +160,7 @@ func (hr *horizontalController) computeScaleAction(
 ) (*datadoghqcommon.DatadogPodAutoscalerHorizontalAction, time.Duration, error) {
 	// Check if we scaling has been disabled explicitly
 	if currentDesiredReplicas == 0 {
-		return nil, 0, errors.New("scaling disabled as current replicas is set to 0")
+		return nil, 0, autoscaling.NewConditionErrorf(autoscaling.ConditionReasonScalingDisabled, "scaling disabled as current replicas is set to 0")
 	}
 
 	// Saving original targetDesiredReplicas
@@ -233,7 +232,7 @@ func (hr *horizontalController) computeScaleAction(
 		if autoscalerInternal.Spec().Fallback != nil && !isFallbackScalingDirectionEnabled(autoscalerInternal.Spec().Fallback.Horizontal.Direction, scaleDirection) {
 			limitReason = fmt.Sprintf("scaling disabled as fallback in the scaling direction (%s) is disabled", scaleDirection)
 			log.Debugf("Scaling limited for autoscaler id: %s, scale direction: %s, limit reason: %s", autoscalerInternal.ID(), scaleDirection, limitReason)
-			return nil, 0, errors.New(limitReason)
+			return nil, 0, autoscaling.NewConditionErrorf(autoscaling.ConditionReasonFallbackRestricted, "%s", limitReason)
 		}
 	}
 
@@ -273,7 +272,7 @@ func (hr *horizontalController) computeScaleAction(
 	allowed, reason := isScalingAllowed(autoscalerSpec, source, scaleDirection)
 	if !allowed {
 		log.Debugf("Scaling not allowed for autoscaler id: %s, scale direction: %s, scale reason: %s (would have scaled to %d replicas)", autoscalerInternal.ID(), scaleDirection, reason, horizontalAction.ToReplicas)
-		return nil, 0, errors.New(reason)
+		return nil, 0, autoscaling.NewConditionErrorf(autoscaling.ConditionReasonPolicyRestricted, "%s", reason)
 	}
 
 	return horizontalAction, evalAfter, nil