updated history init for prometheus to take container types into account

Author: jklaw90

vertical-pod-autoscaler/pkg/recommender/input/cluster_feeder.go

@@ -103,6 +103,7 @@ func (m ClusterStateFeederFactory) Make() *clusterStateFeeder {
 		vpaCheckpointClient: m.VpaCheckpointClient,
 		vpaCheckpointLister: m.VpaCheckpointLister,
 		vpaLister:           m.VpaLister,
+		podLister:           m.PodLister,
 		clusterState:        m.ClusterState,
 		specClient:          spec.NewSpecClient(m.PodLister),
 		selectorFetcher:     m.SelectorFetcher,
@@ -211,6 +212,7 @@ type clusterStateFeeder struct {
 	vpaCheckpointClient vpa_api.VerticalPodAutoscalerCheckpointsGetter
 	vpaCheckpointLister vpa_lister.VerticalPodAutoscalerCheckpointLister
 	vpaLister           vpa_lister.VerticalPodAutoscalerLister
+	podLister           v1lister.PodLister
 	clusterState        model.ClusterState
 	selectorFetcher     target.VpaTargetSelectorFetcher
 	memorySaveMode      bool
@@ -220,55 +222,6 @@ type clusterStateFeeder struct {
 	vpaObjectNamespace  string
 }
 
-func (feeder *clusterStateFeeder) InitFromHistoryProvider(historyProvider history.HistoryProvider) {
-	klog.V(3).InfoS("Initializing VPA from history provider")
-	clusterHistory, err := historyProvider.GetClusterHistory()
-	if err != nil {
-		klog.ErrorS(err, "Cannot get cluster history")
-	}
-	for podID, podHistory := range clusterHistory {
-		klog.V(4).InfoS("Adding pod with labels", "pod", podID, "labels", podHistory.LastLabels)
-		feeder.clusterState.AddOrUpdatePod(podID, podHistory.LastLabels, apiv1.PodUnknown)
-		for containerName, sampleList := range podHistory.Samples {
-			containerID := model.ContainerID{
-				PodID:         podID,
-				ContainerName: containerName,
-			}
-			klog.V(0).InfoS("Adding", "container", containerID)
-			// TODO @jklaw90: pass the container type here
-			if err = feeder.clusterState.AddOrUpdateContainer(containerID, nil, model.ContainerTypeStandard); err != nil {
-				klog.V(0).InfoS("Failed to add container", "container", containerID, "error", err)
-			}
-			klog.V(4).InfoS("Adding samples for container", "sampleCount", len(sampleList), "container", containerID)
-			for _, sample := range sampleList {
-				if err := feeder.clusterState.AddSample(
-					&model.ContainerUsageSampleWithKey{
-						ContainerUsageSample: sample,
-						Container:            containerID,
-					}); err != nil {
-					klog.V(0).InfoS("Failed to add sample", "sample", sample, "error", err)
-				}
-			}
-		}
-	}
-}
-
-func (feeder *clusterStateFeeder) setVpaCheckpoint(checkpoint *vpa_types.VerticalPodAutoscalerCheckpoint) error {
-	vpaID := model.VpaID{Namespace: checkpoint.Namespace, VpaName: checkpoint.Spec.VPAObjectName}
-	vpa, exists := feeder.clusterState.VPAs()[vpaID]
-	if !exists {
-		return fmt.Errorf("cannot load checkpoint to missing VPA object %s/%s", vpaID.Namespace, vpaID.VpaName)
-	}
-
-	cs := model.NewAggregateContainerState()
-	err := cs.LoadFromCheckpoint(&checkpoint.Status)
-	if err != nil {
-		return fmt.Errorf("cannot load checkpoint for VPA %s/%s. Reason: %v", vpaID.Namespace, vpaID.VpaName, err)
-	}
-	vpa.ContainersInitialAggregateState[checkpoint.Spec.ContainerName] = cs
-	return nil
-}
-
 func (feeder *clusterStateFeeder) InitFromCheckpoints(ctx context.Context) {
 	klog.V(3).InfoS("Initializing VPA from checkpoints")
 	feeder.LoadVPAs(ctx)
@@ -300,6 +253,49 @@ func (feeder *clusterStateFeeder) InitFromCheckpoints(ctx context.Context) {
 	}
 }
 
+func (feeder *clusterStateFeeder) InitFromHistoryProvider(historyProvider history.HistoryProvider) {
+	pods := feeder.podSpecLookup()
+	klog.V(3).InfoS("Initializing VPA from history provider")
+	clusterHistory, err := historyProvider.GetClusterHistory()
+	if err != nil {
+		klog.ErrorS(err, "Cannot get cluster history")
+	}
+	for podID, podHistory := range clusterHistory {
+		// no need to load history if the pod no longer exists
+		podSpec, ok := pods[podID]
+		if !ok {
+			continue
+		}
+		klog.V(4).InfoS("Adding pod with labels", "pod", podID, "labels", podHistory.LastLabels)
+		feeder.clusterState.AddOrUpdatePod(podID, podHistory.LastLabels, podSpec.Phase)
+		for containerName, sampleList := range podHistory.Samples {
+			containerID := model.ContainerID{
+				PodID:         podID,
+				ContainerName: containerName,
+			}
+			klog.V(0).InfoS("Adding", "container", containerID)
+
+			containerSpec := podSpec.GetContainerSpec(containerName)
+			if containerSpec == nil {
+				continue
+			}
+			if err = feeder.clusterState.AddOrUpdateContainer(containerID, nil, containerSpec.ContainerType); err != nil {
+				klog.V(0).InfoS("Failed to add container", "container", containerID, "error", err)
+			}
+			klog.V(4).InfoS("Adding samples for container", "sampleCount", len(sampleList), "container", containerID)
+			for _, sample := range sampleList {
+				if err := feeder.clusterState.AddSample(
+					&model.ContainerUsageSampleWithKey{
+						ContainerUsageSample: sample,
+						Container:            containerID,
+					}); err != nil {
+					klog.V(0).InfoS("Failed to add sample", "sample", sample, "error", err)
+				}
+			}
+		}
+	}
+}
+
 func (feeder *clusterStateFeeder) GarbageCollectCheckpoints(ctx context.Context) {
 	klog.V(3).InfoS("Starting garbage collection of checkpoints")
 
@@ -341,82 +337,6 @@ func (feeder *clusterStateFeeder) GarbageCollectCheckpoints(ctx context.Context)
 	}
 }
 
-func (feeder *clusterStateFeeder) shouldIgnoreNamespace(namespace string) bool {
-	// 1. `vpaObjectNamespace` is set but doesn't match the current namespace.
-	if feeder.vpaObjectNamespace != "" && namespace != feeder.vpaObjectNamespace {
-		return true
-	}
-	// 2. `ignoredNamespaces` is set, and the current namespace is in the list.
-	if len(feeder.ignoredNamespaces) > 0 && slices.Contains(feeder.ignoredNamespaces, namespace) {
-		return true
-	}
-	return false
-}
-
-func (feeder *clusterStateFeeder) cleanupCheckpointsForNamespace(ctx context.Context, namespace string, allVPAKeys map[model.VpaID]bool) error {
-	var err error
-	checkpointList, err := feeder.vpaCheckpointLister.VerticalPodAutoscalerCheckpoints(namespace).List(labels.Everything())
-
-	if err != nil {
-		return err
-	}
-	for _, checkpoint := range checkpointList {
-		vpaID := model.VpaID{Namespace: checkpoint.Namespace, VpaName: checkpoint.Spec.VPAObjectName}
-		if !allVPAKeys[vpaID] {
-			if errFeeder := feeder.vpaCheckpointClient.VerticalPodAutoscalerCheckpoints(namespace).Delete(ctx, checkpoint.Name, metav1.DeleteOptions{}); errFeeder != nil {
-				err = fmt.Errorf("failed to delete orphaned checkpoint %s: %w", klog.KRef(namespace, checkpoint.Name), err)
-				continue
-			}
-			klog.V(3).InfoS("Orphaned VPA checkpoint cleanup - deleting", "checkpoint", klog.KRef(namespace, checkpoint.Name))
-		}
-	}
-	return err
-}
-
-func implicitDefaultRecommender(selectors []*vpa_types.VerticalPodAutoscalerRecommenderSelector) bool {
-	return len(selectors) == 0
-}
-
-func selectsRecommender(selectors []*vpa_types.VerticalPodAutoscalerRecommenderSelector, name *string) bool {
-	for _, s := range selectors {
-		if s.Name == *name {
-			return true
-		}
-	}
-	return false
-}
-
-// Filter VPA objects whose specified recommender names are not default
-func filterVPAs(feeder *clusterStateFeeder, allVpaCRDs []*vpa_types.VerticalPodAutoscaler) []*vpa_types.VerticalPodAutoscaler {
-	klog.V(3).InfoS("Start selecting the vpaCRDs.")
-	var vpaCRDs []*vpa_types.VerticalPodAutoscaler
-	for _, vpaCRD := range allVpaCRDs {
-		if feeder.recommenderName == DefaultRecommenderName {
-			if !implicitDefaultRecommender(vpaCRD.Spec.Recommenders) && !selectsRecommender(vpaCRD.Spec.Recommenders, &feeder.recommenderName) {
-				klog.V(6).InfoS("Ignoring vpaCRD as current recommender's name doesn't appear among its recommenders", "vpaCRD", klog.KObj(vpaCRD), "recommenderName", feeder.recommenderName)
-				continue
-			}
-		} else {
-			if implicitDefaultRecommender(vpaCRD.Spec.Recommenders) {
-				klog.V(6).InfoS("Ignoring vpaCRD as recommender doesn't process CRDs implicitly destined to default recommender", "vpaCRD", klog.KObj(vpaCRD), "recommenderName", feeder.recommenderName, "defaultRecommenderName", DefaultRecommenderName)
-				continue
-			}
-			if !selectsRecommender(vpaCRD.Spec.Recommenders, &feeder.recommenderName) {
-				klog.V(6).InfoS("Ignoring vpaCRD as current recommender's name doesn't appear among its recommenders", "vpaCRD", klog.KObj(vpaCRD), "recommenderName", feeder.recommenderName)
-				continue
-			}
-		}
-
-		if feeder.shouldIgnoreNamespace(vpaCRD.Namespace) {
-			klog.V(6).InfoS("Ignoring vpaCRD as this namespace is ignored", "vpaCRD", klog.KObj(vpaCRD))
-			continue
-		}
-
-		vpaCRDs = append(vpaCRDs, vpaCRD)
-	}
-	return vpaCRDs
-}
-
 // LoadVPAs fetches VPA objects and loads them into the cluster state.
 func (feeder *clusterStateFeeder) LoadVPAs(ctx context.Context) {
 	// List VPA API objects.
@@ -468,14 +388,7 @@ func (feeder *clusterStateFeeder) LoadVPAs(ctx context.Context) {
 
 // LoadPods loads pod into the cluster state.
 func (feeder *clusterStateFeeder) LoadPods() {
-	podSpecs, err := feeder.specClient.GetPodSpecs()
-	if err != nil {
-		klog.ErrorS(err, "Cannot get SimplePodSpecs")
-	}
-	pods := make(map[model.PodID]*spec.BasicPodSpec)
-	for _, spec := range podSpecs {
-		pods[spec.ID] = spec
-	}
+	pods := feeder.podSpecLookup()
 	for key := range feeder.clusterState.Pods() {
 		if _, exists := pods[key]; !exists {
 			klog.V(3).InfoS("Deleting Pod", "pod", klog.KRef(key.Namespace, key.PodName))
@@ -488,13 +401,13 @@ func (feeder *clusterStateFeeder) LoadPods() {
 		}
 		feeder.clusterState.AddOrUpdatePod(pod.ID, pod.PodLabels, pod.Phase)
 		for _, container := range pod.Containers {
-			if err = feeder.clusterState.AddOrUpdateContainer(container.ID, container.Request, container.ContainerType); err != nil {
+			if err := feeder.clusterState.AddOrUpdateContainer(container.ID, container.Request, container.ContainerType); err != nil {
 				klog.V(0).InfoS("Failed to add container", "container", container.ID, "error", err)
 			}
 		}
 		for _, initContainer := range pod.InitContainers {
 			if features.Enabled(features.NativeSidecar) && initContainer.ContainerType == model.ContainerTypeInitSidecar {
-				if err = feeder.clusterState.AddOrUpdateContainer(initContainer.ID, initContainer.Request, initContainer.ContainerType); err != nil {
+				if err := feeder.clusterState.AddOrUpdateContainer(initContainer.ID, initContainer.Request, initContainer.ContainerType); err != nil {
 					klog.V(0).InfoS("Failed to add initContainer", "container", initContainer.ID, "error", err)
 				}
 			} else {
@@ -551,6 +464,109 @@ Loop:
 	metrics_recommender.RecordAggregateContainerStatesCount(feeder.clusterState.StateMapSize())
 }
 
+func (feeder *clusterStateFeeder) podSpecLookup() map[model.PodID]*spec.BasicPodSpec {
+	podSpecs, err := feeder.specClient.GetPodSpecs()
+	if err != nil {
+		klog.ErrorS(err, "Cannot get SimplePodSpecs")
+	}
+	pods := make(map[model.PodID]*spec.BasicPodSpec)
+	for _, spec := range podSpecs {
+		pods[spec.ID] = spec
+	}
+	return pods
+}
+
+func (feeder *clusterStateFeeder) setVpaCheckpoint(checkpoint *vpa_types.VerticalPodAutoscalerCheckpoint) error {
+	vpaID := model.VpaID{Namespace: checkpoint.Namespace, VpaName: checkpoint.Spec.VPAObjectName}
+	vpa, exists := feeder.clusterState.VPAs()[vpaID]
+	if !exists {
+		return fmt.Errorf("cannot load checkpoint to missing VPA object %s/%s", vpaID.Namespace, vpaID.VpaName)
+	}
+
+	cs := model.NewAggregateContainerState()
+	err := cs.LoadFromCheckpoint(&checkpoint.Status)
+	if err != nil {
+		return fmt.Errorf("cannot load checkpoint for VPA %s/%s. Reason: %v", vpaID.Namespace, vpaID.VpaName, err)
+	}
+	vpa.ContainersInitialAggregateState[checkpoint.Spec.ContainerName] = cs
+	return nil
+}
+func (feeder *clusterStateFeeder) shouldIgnoreNamespace(namespace string) bool {
+	// 1. `vpaObjectNamespace` is set but doesn't match the current namespace.
+	if feeder.vpaObjectNamespace != "" && namespace != feeder.vpaObjectNamespace {
+		return true
+	}
+	// 2. `ignoredNamespaces` is set, and the current namespace is in the list.
+	if len(feeder.ignoredNamespaces) > 0 && slices.Contains(feeder.ignoredNamespaces, namespace) {
+		return true
+	}
+	return false
+}
+
+func (feeder *clusterStateFeeder) cleanupCheckpointsForNamespace(ctx context.Context, namespace string, allVPAKeys map[model.VpaID]bool) error {
+	var err error
+	checkpointList, err := feeder.vpaCheckpointLister.VerticalPodAutoscalerCheckpoints(namespace).List(labels.Everything())
+
+	if err != nil {
+		return err
+	}
+	for _, checkpoint := range checkpointList {
+		vpaID := model.VpaID{Namespace: checkpoint.Namespace, VpaName: checkpoint.Spec.VPAObjectName}
+		if !allVPAKeys[vpaID] {
+			if errFeeder := feeder.vpaCheckpointClient.VerticalPodAutoscalerCheckpoints(namespace).Delete(ctx, checkpoint.Name, metav1.DeleteOptions{}); errFeeder != nil {
+				err = fmt.Errorf("failed to delete orphaned checkpoint %s: %w", klog.KRef(namespace, checkpoint.Name), err)
+				continue
+			}
+			klog.V(3).InfoS("Orphaned VPA checkpoint cleanup - deleting", "checkpoint", klog.KRef(namespace, checkpoint.Name))
+		}
+	}
+	return err
+}
+
+func implicitDefaultRecommender(selectors []*vpa_types.VerticalPodAutoscalerRecommenderSelector) bool {
+	return len(selectors) == 0
+}
+
+func selectsRecommender(selectors []*vpa_types.VerticalPodAutoscalerRecommenderSelector, name *string) bool {
+	for _, s := range selectors {
+		if s.Name == *name {
+			return true
+		}
+	}
+	return false
+}
+
+// Filter VPA objects whose specified recommender names are not default
+func filterVPAs(feeder *clusterStateFeeder, allVpaCRDs []*vpa_types.VerticalPodAutoscaler) []*vpa_types.VerticalPodAutoscaler {
+	klog.V(3).InfoS("Start selecting the vpaCRDs.")
+	var vpaCRDs []*vpa_types.VerticalPodAutoscaler
+	for _, vpaCRD := range allVpaCRDs {
+		if feeder.recommenderName == DefaultRecommenderName {
+			if !implicitDefaultRecommender(vpaCRD.Spec.Recommenders) && !selectsRecommender(vpaCRD.Spec.Recommenders, &feeder.recommenderName) {
+				klog.V(6).InfoS("Ignoring vpaCRD as current recommender's name doesn't appear among its recommenders", "vpaCRD", klog.KObj(vpaCRD), "recommenderName", feeder.recommenderName)
+				continue
+			}
+		} else {
+			if implicitDefaultRecommender(vpaCRD.Spec.Recommenders) {
+				klog.V(6).InfoS("Ignoring vpaCRD as recommender doesn't process CRDs implicitly destined to default recommender", "vpaCRD", klog.KObj(vpaCRD), "recommenderName", feeder.recommenderName, "defaultRecommenderName", DefaultRecommenderName)
+				continue
+			}
+			if !selectsRecommender(vpaCRD.Spec.Recommenders, &feeder.recommenderName) {
+				klog.V(6).InfoS("Ignoring vpaCRD as current recommender's name doesn't appear among its recommenders", "vpaCRD", klog.KObj(vpaCRD), "recommenderName", feeder.recommenderName)
+				continue
+			}
+		}
+
+		if feeder.shouldIgnoreNamespace(vpaCRD.Namespace) {
+			klog.V(6).InfoS("Ignoring vpaCRD as this namespace is ignored", "vpaCRD", klog.KObj(vpaCRD))
+			continue
+		}
+
+		vpaCRDs = append(vpaCRDs, vpaCRD)
+	}
+	return vpaCRDs
+}
+
 func (feeder *clusterStateFeeder) matchesVPA(pod *spec.BasicPodSpec) bool {
 	for vpaKey, vpa := range feeder.clusterState.VPAs() {
 		podLabels := labels.Set(pod.PodLabels)