Add capacity buffers scale up events

Author: abdelrahman882

cluster-autoscaler/main.go

@@ -192,9 +192,10 @@ func buildAutoscaler(context ctx.Context, debuggingSnapshotter debuggingsnapshot
 			capacitybufferClient, capacitybufferClientError = capacityclient.NewCapacityBufferClientFromConfig(restConfig)
 		}
 		if capacitybufferClientError == nil && capacitybufferClient != nil {
-			bufferPodInjector := cbprocessor.NewCapacityBufferPodListProcessor(capacitybufferClient, []string{common.ActiveProvisioningStrategy})
+			buffersPodsRegistry := cbprocessor.NewDefaultCapacityBuffersFakePodsRegistry()
+			bufferPodInjector := cbprocessor.NewCapacityBufferPodListProcessor(capacitybufferClient, []string{common.ActiveProvisioningStrategy}, buffersPodsRegistry)
 			podListProcessor = pods.NewCombinedPodListProcessor([]pods.PodListProcessor{bufferPodInjector, podListProcessor})
-			opts.Processors.ScaleUpStatusProcessor = status.NewCombinedScaleUpStatusProcessor([]status.ScaleUpStatusProcessor{cbprocessor.NewFakePodsScaleUpStatusProcessor(), opts.Processors.ScaleUpStatusProcessor})
+			opts.Processors.ScaleUpStatusProcessor = status.NewCombinedScaleUpStatusProcessor([]status.ScaleUpStatusProcessor{cbprocessor.NewFakePodsScaleUpStatusProcessor(buffersPodsRegistry), opts.Processors.ScaleUpStatusProcessor})
 		}
 	}
 

cluster-autoscaler/processors/capacitybuffer/pod_list_processor.go

@@ -26,7 +26,7 @@ import (
 	"k8s.io/klog/v2"
 
 	apiv1 "k8s.io/api/core/v1"
-	api_v1 "k8s.io/autoscaler/cluster-autoscaler/apis/capacitybuffer/autoscaling.x-k8s.io/v1alpha1"
+	v1alpha1 "k8s.io/autoscaler/cluster-autoscaler/apis/capacitybuffer/autoscaling.x-k8s.io/v1alpha1"
 	client "k8s.io/autoscaler/cluster-autoscaler/capacitybuffer/client"
 	"k8s.io/autoscaler/cluster-autoscaler/capacitybuffer/common"
 	buffersfilter "k8s.io/autoscaler/cluster-autoscaler/capacitybuffer/filters"
@@ -35,8 +35,8 @@ import (
 
 // Pods annotation keys and values for fake pods created by capacity buffer pod list processor
 const (
-	FakeCapacityBufferPodAnnotationKey   = "podType"
-	FakeCapacityBufferPodAnnotationValue = "capacityBufferFakePod"
+	CapacityBufferFakePodAnnotationKey   = "podType"
+	CapacityBufferFakePodAnnotationValue = "capacityBufferFakePod"
 )
 
 // CapacityBufferPodListProcessor processes the pod lists before scale up
@@ -46,10 +46,27 @@ type CapacityBufferPodListProcessor struct {
 	statusFilter         buffersfilter.Filter
 	podTemplateGenFilter buffersfilter.Filter
 	provStrategies       map[string]bool
+	buffersRegistry      *capacityBuffersFakePodsRegistry
+}
+
+// capacityBuffersFakePodsRegistry a struct that keeps the status of capacity buffer
+// the fake pods generated for adding buffer event later
+type capacityBuffersFakePodsRegistry struct {
+	fakePodsUIDToBuffer map[string]*v1alpha1.CapacityBuffer
+}
+
+// NewCapacityBuffersFakePodsRegistry returns a new pointer to empty capacityBuffersFakePodsRegistry
+func NewCapacityBuffersFakePodsRegistry(fakePodsToBuffers map[string]*v1alpha1.CapacityBuffer) *capacityBuffersFakePodsRegistry {
+	return &capacityBuffersFakePodsRegistry{fakePodsUIDToBuffer: fakePodsToBuffers}
+}
+
+// NewDefaultCapacityBuffersFakePodsRegistry returns a new pointer to empty capacityBuffersFakePodsRegistry
+func NewDefaultCapacityBuffersFakePodsRegistry() *capacityBuffersFakePodsRegistry {
+	return &capacityBuffersFakePodsRegistry{fakePodsUIDToBuffer: map[string]*v1alpha1.CapacityBuffer{}}
 }
 
 // NewCapacityBufferPodListProcessor creates a new CapacityRequestPodListProcessor.
-func NewCapacityBufferPodListProcessor(client *client.CapacityBufferClient, provStrategies []string) *CapacityBufferPodListProcessor {
+func NewCapacityBufferPodListProcessor(client *client.CapacityBufferClient, provStrategies []string, buffersRegistry *capacityBuffersFakePodsRegistry) *CapacityBufferPodListProcessor {
 	provStrategiesMap := map[string]bool{}
 	for _, ps := range provStrategies {
 		provStrategiesMap[ps] = true
@@ -62,6 +79,7 @@ func NewCapacityBufferPodListProcessor(client *client.CapacityBufferClient, prov
 		}),
 		podTemplateGenFilter: buffersfilter.NewPodTemplateGenerationChangedFilter(client),
 		provStrategies:       provStrategiesMap,
+		buffersRegistry:      buffersRegistry,
 	}
 }
 
@@ -79,6 +97,7 @@ func (p *CapacityBufferPodListProcessor) Process(ctx *context.AutoscalingContext
 	totalFakePods := []*apiv1.Pod{}
 	for _, buffer := range buffers {
 		fakePods := p.provision(buffer)
+		p.updateCapacityBufferRegistry(fakePods, buffer)
 		totalFakePods = append(totalFakePods, fakePods...)
 	}
 	klog.V(2).Infof("Capacity pod processor injecting %v fake pods provisioning %v capacity buffers", len(totalFakePods), len(buffers))
@@ -90,7 +109,16 @@ func (p *CapacityBufferPodListProcessor) Process(ctx *context.AutoscalingContext
 func (p *CapacityBufferPodListProcessor) CleanUp() {
 }
 
-func (p *CapacityBufferPodListProcessor) provision(buffer *api_v1.CapacityBuffer) []*apiv1.Pod {
+func (p *CapacityBufferPodListProcessor) updateCapacityBufferRegistry(fakePods []*apiv1.Pod, buffer *v1alpha1.CapacityBuffer) {
+	p.buffersRegistry.fakePodsUIDToBuffer = make(map[string]*v1alpha1.CapacityBuffer, len(fakePods))
+	for _, fakePod := range fakePods {
+		if p.buffersRegistry != nil {
+			p.buffersRegistry.fakePodsUIDToBuffer[string(fakePod.UID)] = buffer
+		}
+	}
+}
+
+func (p *CapacityBufferPodListProcessor) provision(buffer *v1alpha1.CapacityBuffer) []*apiv1.Pod {
 	if buffer.Status.PodTemplateRef == nil || buffer.Status.Replicas == nil {
 		return []*apiv1.Pod{}
 	}
@@ -102,7 +130,7 @@ func (p *CapacityBufferPodListProcessor) provision(buffer *api_v1.CapacityBuffer
 		p.updateBufferStatus(buffer)
 		return []*apiv1.Pod{}
 	}
-	fakePods, err := makeFakePods(buffer.Name, &podTemplate.Template, int(*replicas))
+	fakePods, err := makeFakePods(buffer, &podTemplate.Template, int(*replicas))
 	if err != nil {
 		common.UpdateBufferStatusToFailedProvisioing(buffer, "FailedToMakeFakePods", fmt.Sprintf("failed to create fake pods with error: %v", err.Error()))
 		p.updateBufferStatus(buffer)
@@ -113,8 +141,8 @@ func (p *CapacityBufferPodListProcessor) provision(buffer *api_v1.CapacityBuffer
 	return fakePods
 }
 
-func (p *CapacityBufferPodListProcessor) filterBuffersProvStrategy(buffers []*api_v1.CapacityBuffer) []*api_v1.CapacityBuffer {
-	var filteredBuffers []*api_v1.CapacityBuffer
+func (p *CapacityBufferPodListProcessor) filterBuffersProvStrategy(buffers []*v1alpha1.CapacityBuffer) []*v1alpha1.CapacityBuffer {
+	var filteredBuffers []*v1alpha1.CapacityBuffer
 	for _, buffer := range buffers {
 
 		if buffer.Status.ProvisioningStrategy != nil && p.provStrategies[*buffer.Status.ProvisioningStrategy] {
@@ -124,24 +152,24 @@ func (p *CapacityBufferPodListProcessor) filterBuffersProvStrategy(buffers []*ap
 	return filteredBuffers
 }
 
-func (p *CapacityBufferPodListProcessor) updateBufferStatus(buffer *api_v1.CapacityBuffer) {
+func (p *CapacityBufferPodListProcessor) updateBufferStatus(buffer *v1alpha1.CapacityBuffer) {
 	_, err := p.client.UpdateCapacityBuffer(buffer)
 	if err != nil {
 		klog.Errorf("Failed to update buffer status for buffer %v, error: %v", buffer.Name, err.Error())
 	}
 }
 
 // makeFakePods creates podCount number of copies of the sample pod
-func makeFakePods(bufferName string, samplePodTemplate *apiv1.PodTemplateSpec, podCount int) ([]*apiv1.Pod, error) {
+func makeFakePods(buffer *v1alpha1.CapacityBuffer, samplePodTemplate *apiv1.PodTemplateSpec, podCount int) ([]*apiv1.Pod, error) {
 	var fakePods []*apiv1.Pod
 	samplePod := getPodFromTemplate(samplePodTemplate)
-	samplePod = withCapacityBufferFakePodAnnotation(samplePod)
 	for i := 1; i <= podCount; i++ {
-		newPod := samplePod.DeepCopy()
-		newPod.Name = fmt.Sprintf("capacity-buffer-%s-%d", bufferName, i)
-		newPod.UID = types.UID(fmt.Sprintf("%s-%d", string(bufferName), i))
-		newPod.Spec.NodeName = ""
-		fakePods = append(fakePods, newPod)
+		fakePod := samplePod.DeepCopy()
+		fakePod = withCapacityBufferFakePodAnnotation(fakePod)
+		fakePod.Name = fmt.Sprintf("capacity-buffer-%s-%d", buffer.Name, i)
+		fakePod.UID = types.UID(fmt.Sprintf("%s-%d", string(buffer.UID), i))
+		fakePod.Spec.NodeName = ""
+		fakePods = append(fakePods, fakePod)
 	}
 	return fakePods, nil
 }
@@ -150,15 +178,15 @@ func withCapacityBufferFakePodAnnotation(pod *apiv1.Pod) *apiv1.Pod {
 	if pod.Annotations == nil {
 		pod.Annotations = make(map[string]string, 1)
 	}
-	pod.Annotations[FakeCapacityBufferPodAnnotationKey] = FakeCapacityBufferPodAnnotationValue
+	pod.Annotations[CapacityBufferFakePodAnnotationKey] = CapacityBufferFakePodAnnotationValue
 	return pod
 }
 
 func isFakeCapacityBuffersPod(pod *apiv1.Pod) bool {
 	if pod.Annotations == nil {
 		return false
 	}
-	return pod.Annotations[FakeCapacityBufferPodAnnotationKey] == FakeCapacityBufferPodAnnotationValue
+	return pod.Annotations[CapacityBufferFakePodAnnotationKey] == CapacityBufferFakePodAnnotationValue
 }
 
 func getPodFromTemplate(template *apiv1.PodTemplateSpec) *apiv1.Pod {

cluster-autoscaler/processors/capacitybuffer/pod_list_processor_test.go

@@ -143,7 +143,7 @@ func TestPodListProcessor(t *testing.T) {
 			fakeBuffersClient := buffersfake.NewSimpleClientset(test.objectsInBuffersClient...)
 			fakeCapacityBuffersClient, _ := client.NewCapacityBufferClientFromClients(fakeBuffersClient, fakeKubernetesClient, nil, nil)
 
-			processor := NewCapacityBufferPodListProcessor(fakeCapacityBuffersClient, []string{testProvStrategyAllowed})
+			processor := NewCapacityBufferPodListProcessor(fakeCapacityBuffersClient, []string{testProvStrategyAllowed}, NewDefaultCapacityBuffersFakePodsRegistry())
 			resUnschedulablePods, err := processor.Process(nil, test.unschedulablePods)
 			assert.Equal(t, err != nil, test.expectError)
 

cluster-autoscaler/processors/capacitybuffer/scale_up_status_processor.go

@@ -17,46 +17,124 @@ limitations under the License.
 package capacitybufferpodlister
 
 import (
+	"fmt"
+	"strings"
+
 	apiv1 "k8s.io/api/core/v1"
+	v1alpha1 "k8s.io/autoscaler/cluster-autoscaler/apis/capacitybuffer/autoscaling.x-k8s.io/v1alpha1"
 	ca_context "k8s.io/autoscaler/cluster-autoscaler/context"
 	"k8s.io/autoscaler/cluster-autoscaler/processors/status"
-	"k8s.io/klog/v2"
 )
 
 // FakePodsScaleUpStatusProcessor is a ScaleUpStatusProcessor used for filtering out fake pods from scaleup status.
-type FakePodsScaleUpStatusProcessor struct{}
+type FakePodsScaleUpStatusProcessor struct {
+	buffersRegistry *capacityBuffersFakePodsRegistry
+}
+
+type bufferInfo struct {
+	buffer         *v1alpha1.CapacityBuffer
+	numberOfPods   int
+	reasonMessages []string
+}
 
 // NewFakePodsScaleUpStatusProcessor return an instance of FakePodsScaleUpStatusProcessor
-func NewFakePodsScaleUpStatusProcessor() *FakePodsScaleUpStatusProcessor {
-	return &FakePodsScaleUpStatusProcessor{}
+func NewFakePodsScaleUpStatusProcessor(buffersRegistry *capacityBuffersFakePodsRegistry) *FakePodsScaleUpStatusProcessor {
+	return &FakePodsScaleUpStatusProcessor{buffersRegistry: buffersRegistry}
 }
 
 // Process updates scaleupStatus to remove all capacity buffer fake pods from
 // PodsRemainUnschedulable, PodsAwaitEvaluation & PodsTriggeredScaleup
-func (a *FakePodsScaleUpStatusProcessor) Process(_ *ca_context.AutoscalingContext, scaleUpStatus *status.ScaleUpStatus) {
-	scaleUpStatus.PodsRemainUnschedulable = filterFakePods(scaleUpStatus.PodsRemainUnschedulable, func(noScaleUpInfo status.NoScaleUpInfo) *apiv1.Pod { return noScaleUpInfo.Pod }, "PodsRemainUnschedulable")
-	scaleUpStatus.PodsAwaitEvaluation = filterFakePods(scaleUpStatus.PodsAwaitEvaluation, func(pod *apiv1.Pod) *apiv1.Pod { return pod }, "PodsAwaitEvaluation")
-	scaleUpStatus.PodsTriggeredScaleUp = filterFakePods(scaleUpStatus.PodsTriggeredScaleUp, func(pod *apiv1.Pod) *apiv1.Pod { return pod }, "PodsTriggeredScaleUp")
+func (p *FakePodsScaleUpStatusProcessor) Process(context *ca_context.AutoscalingContext, scaleUpStatus *status.ScaleUpStatus) {
+	var fakePodsRemainUnschedulable []status.NoScaleUpInfo
+	var fakePodsTriggeredScaleUp []*apiv1.Pod
+
+	scaleUpStatus.PodsRemainUnschedulable, fakePodsRemainUnschedulable = filterOutFakeUnschedulablePods(scaleUpStatus.PodsRemainUnschedulable)
+	scaleUpStatus.PodsTriggeredScaleUp, fakePodsTriggeredScaleUp = filterOutFakePods(scaleUpStatus.PodsTriggeredScaleUp)
+	scaleUpStatus.PodsAwaitEvaluation, _ = filterOutFakePods(scaleUpStatus.PodsAwaitEvaluation)
+
+	p.createEventsForBuffersWithUnschedulablePods(context, scaleUpStatus, fakePodsRemainUnschedulable)
+	p.createEventsForBuffersWithPodsTriggeredScaleUp(context, scaleUpStatus, fakePodsTriggeredScaleUp)
+}
+
+func (p *FakePodsScaleUpStatusProcessor) createEventsForBuffersWithUnschedulablePods(context *ca_context.AutoscalingContext, scaleUpStatus *status.ScaleUpStatus, fakePodsRemainUnschedulable []status.NoScaleUpInfo) {
+	if scaleUpStatus.Result != status.ScaleUpSuccessful && scaleUpStatus.Result != status.ScaleUpError {
+		consideredNodeGroupsMap := status.NodeGroupListToMapById(scaleUpStatus.ConsideredNodeGroups)
+		buffersInfo := map[string]*bufferInfo{}
+		for _, noScaleUpInfo := range fakePodsRemainUnschedulable {
+			parentCapacityBuffer, found := p.buffersRegistry.fakePodsUIDToBuffer[string(noScaleUpInfo.Pod.UID)]
+			if found {
+				bufferUID := string(parentCapacityBuffer.UID)
+				if _, found := buffersInfo[bufferUID]; !found {
+					buffersInfo[bufferUID] = &bufferInfo{
+						buffer: parentCapacityBuffer,
+					}
+				}
+				buffersInfo[bufferUID].numberOfPods += 1
+				buffersInfo[bufferUID].reasonMessages = append(buffersInfo[bufferUID].reasonMessages,
+					status.ReasonsMessage(scaleUpStatus.Result, noScaleUpInfo, consideredNodeGroupsMap))
+			}
+		}
+
+		for _, bufferInfo := range buffersInfo {
+			context.Recorder.Event(bufferInfo.buffer, apiv1.EventTypeNormal, "NotTriggerScaleUp",
+				fmt.Sprintf("capacity buffer %d fake pods didn't trigger scale-up: %s",
+					bufferInfo.numberOfPods, strings.Join(bufferInfo.reasonMessages, ",")))
+		}
+	}
+}
+
+func (p *FakePodsScaleUpStatusProcessor) createEventsForBuffersWithPodsTriggeredScaleUp(context *ca_context.AutoscalingContext, scaleUpStatus *status.ScaleUpStatus, fakePodsTriggeredScaleUp []*apiv1.Pod) {
+	if len(scaleUpStatus.ScaleUpInfos) > 0 && len(fakePodsTriggeredScaleUp) > 0 {
+		buffersInfo := map[string]*bufferInfo{}
+		for _, pod := range fakePodsTriggeredScaleUp {
+			parentCapacityBuffer, found := p.buffersRegistry.fakePodsUIDToBuffer[string(pod.UID)]
+			if found {
+				bufferUID := string(parentCapacityBuffer.UID)
+				if _, found := buffersInfo[bufferUID]; !found {
+					buffersInfo[bufferUID] = &bufferInfo{
+						buffer: parentCapacityBuffer,
+					}
+				}
+				buffersInfo[bufferUID].numberOfPods += 1
+			}
+		}
+		for _, bufferInfo := range buffersInfo {
+			context.Recorder.Eventf(bufferInfo.buffer, apiv1.EventTypeNormal, "TriggeredScaleUp",
+				"capacity buffer %d fake pods triggered scale-up: %v", bufferInfo.numberOfPods, scaleUpStatus.ScaleUpInfos)
+		}
+	}
 }
 
-// filterFakePods removes capacity buffer fake pods from the input list of T using passed getPod(T)
-// Returns a list containing only non-fake pods
-func filterFakePods[T any](podsWrappers []T, getPod func(T) *apiv1.Pod, resourceName string) []T {
-	filteredPodsSources := make([]T, 0)
-	removedPods := make([]*apiv1.Pod, 0)
+// filterOutFakeUnschedulablePods filters out NoScaleUpInfo for capacity buffers fake pods
+func filterOutFakeUnschedulablePods(noScaleUpInfo []status.NoScaleUpInfo) ([]status.NoScaleUpInfo, []status.NoScaleUpInfo) {
+	filteredInfo := make([]status.NoScaleUpInfo, 0)
+	filteredOutInfo := make([]status.NoScaleUpInfo, 0)
 
-	for _, podsWrapper := range podsWrappers {
-		currentPod := getPod(podsWrapper)
+	for _, info := range noScaleUpInfo {
+		currentPod := info.Pod
 		if !isFakeCapacityBuffersPod(currentPod) {
-			filteredPodsSources = append(filteredPodsSources, podsWrapper)
+			filteredInfo = append(filteredInfo, info)
 			continue
 		}
-		removedPods = append(removedPods, currentPod)
+		filteredOutInfo = append(filteredOutInfo, info)
 	}
+	return filteredInfo, filteredOutInfo
+}
+
+// filterFakePods filters out capacity buffer fake pods
+func filterOutFakePods(pods []*apiv1.Pod) ([]*apiv1.Pod, []*apiv1.Pod) {
+	filteredPods := make([]*apiv1.Pod, 0)
+	filteredOutPods := make([]*apiv1.Pod, 0)
 
-	klog.Infof("Filtered out %d pods from %s", len(removedPods), resourceName)
-	return filteredPodsSources
+	for _, pod := range pods {
+		if !isFakeCapacityBuffersPod(pod) {
+			filteredPods = append(filteredPods, pod)
+			continue
+		}
+		filteredOutPods = append(filteredOutPods, pod)
+	}
+	return filteredPods, filteredOutPods
 }
 
 // CleanUp is called at CA termination
-func (a *FakePodsScaleUpStatusProcessor) CleanUp() {}
+func (p *FakePodsScaleUpStatusProcessor) CleanUp() {}