Merge branch 'master' of github.com:kgolab/autoscaler into vpa-updater-metrics

Author: kgolab

cluster-autoscaler/cloudprovider/clusterapi/clusterapi_controller_test.go

@@ -987,7 +987,7 @@ func TestControllerMachineSetNodeNamesUsingProviderID(t *testing.T) {
 	})
 
 	for i := range testConfig.nodes {
-		if nodeNames[i].Id != string(normalizedProviderString(testConfig.nodes[i].Spec.ProviderID)) {
+		if nodeNames[i].Id != testConfig.nodes[i].Spec.ProviderID {
 			t.Fatalf("expected %q, got %q", testConfig.nodes[i].Spec.ProviderID, nodeNames[i].Id)
 		}
 	}
@@ -1035,7 +1035,7 @@ func TestControllerMachineSetNodeNamesUsingStatusNodeRefName(t *testing.T) {
 	})
 
 	for i := range testConfig.nodes {
-		if nodeNames[i].Id != string(normalizedProviderString(testConfig.nodes[i].Spec.ProviderID)) {
+		if nodeNames[i].Id != testConfig.nodes[i].Spec.ProviderID {
 			t.Fatalf("expected %q, got %q", testConfig.nodes[i].Spec.ProviderID, nodeNames[i].Id)
 		}
 	}

cluster-autoscaler/cloudprovider/clusterapi/clusterapi_nodegroup.go

@@ -195,10 +195,14 @@ func (ng *nodegroup) Nodes() ([]cloudprovider.Instance, error) {
 		return nil, err
 	}
 
+	// Nodes do not have normalized IDs, so do not normalize the ID here.
+	// The IDs returned here are used to check if a node is registered or not and
+	// must match the ID on the Node object itself.
+	// https://github.com/kubernetes/autoscaler/blob/a973259f1852303ba38a3a61eeee8489cf4e1b13/cluster-autoscaler/clusterstate/clusterstate.go#L967-L985
 	instances := make([]cloudprovider.Instance, len(nodes))
 	for i := range nodes {
 		instances[i] = cloudprovider.Instance{
-			Id: string(normalizedProviderString(nodes[i])),
+			Id: nodes[i],
 		}
 	}
 

cluster-autoscaler/cloudprovider/clusterapi/clusterapi_nodegroup_test.go

@@ -659,7 +659,7 @@ func TestNodeGroupDeleteNodes(t *testing.T) {
 		})
 
 		for i := 0; i < len(nodeNames); i++ {
-			if nodeNames[i].Id != string(normalizedProviderString(testConfig.nodes[i].Spec.ProviderID)) {
+			if nodeNames[i].Id != testConfig.nodes[i].Spec.ProviderID {
 				t.Fatalf("expected %q, got %q", testConfig.nodes[i].Spec.ProviderID, nodeNames[i].Id)
 			}
 		}
@@ -844,7 +844,7 @@ func TestNodeGroupDeleteNodesTwice(t *testing.T) {
 		})
 
 		for i := 0; i < len(nodeNames); i++ {
-			if nodeNames[i].Id != string(normalizedProviderString(testConfig.nodes[i].Spec.ProviderID)) {
+			if nodeNames[i].Id != testConfig.nodes[i].Spec.ProviderID {
 				t.Fatalf("expected %q, got %q", testConfig.nodes[i].Spec.ProviderID, nodeNames[i].Id)
 			}
 		}
@@ -998,7 +998,7 @@ func TestNodeGroupWithFailedMachine(t *testing.T) {
 				nodeIndex = i
 			}
 
-			if nodeNames[i].Id != string(normalizedProviderString(testConfig.nodes[nodeIndex].Spec.ProviderID)) {
+			if nodeNames[i].Id != testConfig.nodes[nodeIndex].Spec.ProviderID {
 				t.Fatalf("expected %q, got %q", testConfig.nodes[nodeIndex].Spec.ProviderID, nodeNames[i].Id)
 			}
 		}

cluster-autoscaler/core/scale_up.go

@@ -464,6 +464,7 @@ func ScaleUp(context *context.AutoscalingContext, processors *ca_processors.Auto
 		if context.MaxNodesTotal > 0 && len(nodes)+newNodes+len(upcomingNodes) > context.MaxNodesTotal {
 			klog.V(1).Infof("Capping size to max cluster total size (%d)", context.MaxNodesTotal)
 			newNodes = context.MaxNodesTotal - len(nodes) - len(upcomingNodes)
+			context.LogRecorder.Eventf(apiv1.EventTypeWarning, "MaxNodesTotalReached", "Max total nodes in cluster reached: %v", context.MaxNodesTotal)
 			if newNodes < 1 {
 				return &status.ScaleUpStatus{Result: status.ScaleUpError}, errors.NewAutoscalerError(
 					errors.TransientError,

vertical-pod-autoscaler/deploy/vpa-rbac.yaml

@@ -156,6 +156,13 @@ kind: ClusterRole
 metadata:
   name: system:vpa-target-reader
 rules:
+  - apiGroups:
+    - '*'
+    resources:
+    - '*/scale'
+    verbs:
+    - get
+    - watch
   - apiGroups:
       - ""
     resources:

vertical-pod-autoscaler/pkg/recommender/model/cluster.go

@@ -46,9 +46,8 @@ type ClusterState struct {
 	// time we've noticed the recommendation missing or last time we logged
 	// a warning about it.
 	EmptyVPAs map[VpaID]time.Time
-	// VpasWithMatchingPods contains information if there exist live pods that
-	// this VPAs selector matches.
-	VpasWithMatchingPods map[VpaID]bool
+	// VpaPodCount contains number of live Pods matching a given VPA object.
+	VpaPodCount map[VpaID]int
 	// Observed VPAs. Used to check if there are updates needed.
 	ObservedVpas []*vpa_types.VerticalPodAutoscaler
 
@@ -97,12 +96,12 @@ type PodState struct {
 // NewClusterState returns a new ClusterState with no pods.
 func NewClusterState() *ClusterState {
 	return &ClusterState{
-		Pods:                 make(map[PodID]*PodState),
-		Vpas:                 make(map[VpaID]*Vpa),
-		EmptyVPAs:            make(map[VpaID]time.Time),
-		VpasWithMatchingPods: make(map[VpaID]bool),
-		aggregateStateMap:    make(aggregateContainerStatesMap),
-		labelSetMap:          make(labelSetMap),
+		Pods:              make(map[PodID]*PodState),
+		Vpas:              make(map[VpaID]*Vpa),
+		EmptyVPAs:         make(map[VpaID]time.Time),
+		VpaPodCount:       make(map[VpaID]int),
+		aggregateStateMap: make(aggregateContainerStatesMap),
+		labelSetMap:       make(labelSetMap),
 	}
 }
 
@@ -124,18 +123,44 @@ func (cluster *ClusterState) AddOrUpdatePod(podID PodID, newLabels labels.Set, p
 		pod = newPod(podID)
 		cluster.Pods[podID] = pod
 	}
+
 	newlabelSetKey := cluster.getLabelSetKey(newLabels)
+	if podExists && pod.labelSetKey != newlabelSetKey {
+		// This Pod is already counted in the old VPA, remove the link.
+		cluster.removePodFromItsVpa(pod)
+	}
 	if !podExists || pod.labelSetKey != newlabelSetKey {
 		pod.labelSetKey = newlabelSetKey
 		// Set the links between the containers and aggregations based on the current pod labels.
 		for containerName, container := range pod.Containers {
 			containerID := ContainerID{PodID: podID, ContainerName: containerName}
 			container.aggregator = cluster.findOrCreateAggregateContainerState(containerID)
 		}
+
+		cluster.addPodToItsVpa(pod)
 	}
 	pod.Phase = phase
 }
 
+// addPodToItsVpa increases the count of Pods associated with a VPA object.
+// Does a scan similar to findOrCreateAggregateContainerState so could be optimized if needed.
+func (cluster *ClusterState) addPodToItsVpa(pod *PodState) {
+	for _, vpa := range cluster.Vpas {
+		if vpa_utils.PodLabelsMatchVPA(pod.ID.Namespace, cluster.labelSetMap[pod.labelSetKey], vpa.ID.Namespace, vpa.PodSelector) {
+			cluster.VpaPodCount[vpa.ID]++
+		}
+	}
+}
+
+// removePodFromItsVpa decreases the count of Pods associated with a VPA object.
+func (cluster *ClusterState) removePodFromItsVpa(pod *PodState) {
+	for _, vpa := range cluster.Vpas {
+		if vpa_utils.PodLabelsMatchVPA(pod.ID.Namespace, cluster.labelSetMap[pod.labelSetKey], vpa.ID.Namespace, vpa.PodSelector) {
+			cluster.VpaPodCount[vpa.ID]--
+		}
+	}
+}
+
 // GetContainer returns the ContainerState object for a given ContainerID or
 // null if it's not present in the model.
 func (cluster *ClusterState) GetContainer(containerID ContainerID) *ContainerState {
@@ -151,6 +176,10 @@ func (cluster *ClusterState) GetContainer(containerID ContainerID) *ContainerSta
 
 // DeletePod removes an existing pod from the cluster.
 func (cluster *ClusterState) DeletePod(podID PodID) {
+	pod, found := cluster.Pods[podID]
+	if found {
+		cluster.removePodFromItsVpa(pod)
+	}
 	delete(cluster.Pods, podID)
 }
 
@@ -237,10 +266,9 @@ func (cluster *ClusterState) AddOrUpdateVpa(apiObject *vpa_types.VerticalPodAuto
 		vpa = NewVpa(vpaID, selector, apiObject.CreationTimestamp.Time)
 		cluster.Vpas[vpaID] = vpa
 		for aggregationKey, aggregation := range cluster.aggregateStateMap {
-			if vpa.UseAggregationIfMatching(aggregationKey, aggregation) {
-				cluster.VpasWithMatchingPods[vpa.ID] = true
-			}
+			vpa.UseAggregationIfMatching(aggregationKey, aggregation)
 		}
+		cluster.VpaPodCount[vpaID] = len(cluster.GetMatchingPods(vpa))
 	}
 	vpa.TargetRef = apiObject.Spec.TargetRef
 	vpa.Annotations = annotationsMap
@@ -262,7 +290,7 @@ func (cluster *ClusterState) DeleteVpa(vpaID VpaID) error {
 	}
 	delete(cluster.Vpas, vpaID)
 	delete(cluster.EmptyVPAs, vpaID)
-	delete(cluster.VpasWithMatchingPods, vpaID)
+	delete(cluster.VpaPodCount, vpaID)
 	return nil
 }
 
@@ -313,9 +341,7 @@ func (cluster *ClusterState) findOrCreateAggregateContainerState(containerID Con
 		cluster.aggregateStateMap[aggregateStateKey] = aggregateContainerState
 		// Link the new aggregation to the existing VPAs.
 		for _, vpa := range cluster.Vpas {
-			if vpa.UseAggregationIfMatching(aggregateStateKey, aggregateContainerState) {
-				cluster.VpasWithMatchingPods[vpa.ID] = true
-			}
+			vpa.UseAggregationIfMatching(aggregateStateKey, aggregateContainerState)
 		}
 	}
 	return aggregateContainerState

vertical-pod-autoscaler/pkg/recommender/model/cluster_test.go

@@ -32,6 +32,8 @@ import (
 
 var (
 	testPodID       = PodID{"namespace-1", "pod-1"}
+	testPodID3      = PodID{"namespace-1", "pod-3"}
+	testPodID4      = PodID{"namespace-1", "pod-4"}
 	testContainerID = ContainerID{testPodID, "container-1"}
 	testVpaID       = VpaID{"namespace-1", "vpa-1"}
 	testAnnotations = vpaAnnotationsMap{"key-1": "value-1"}
@@ -701,13 +703,13 @@ func TestVPAWithMatchingPods(t *testing.T) {
 		name          string
 		vpaSelector   string
 		pods          []podDesc
-		expectedMatch bool
+		expectedMatch int
 	}{
 		{
 			name:          "No pods",
 			vpaSelector:   testSelectorStr,
 			pods:          []podDesc{},
-			expectedMatch: false,
+			expectedMatch: 0,
 		},
 		{
 			name:        "VPA with matching pod",
@@ -719,7 +721,7 @@ func TestVPAWithMatchingPods(t *testing.T) {
 					apiv1.PodRunning,
 				},
 			},
-			expectedMatch: true,
+			expectedMatch: 1,
 		},
 		{
 			name:        "No matching pod",
@@ -731,19 +733,55 @@ func TestVPAWithMatchingPods(t *testing.T) {
 					apiv1.PodRunning,
 				},
 			},
-			expectedMatch: false,
+			expectedMatch: 0,
+		},
+		{
+			name:        "VPA with 2 matching pods, 1 not matching",
+			vpaSelector: testSelectorStr,
+			pods: []podDesc{
+				{
+					testPodID,
+					emptyLabels, // does not match VPA
+					apiv1.PodRunning,
+				},
+				{
+					testPodID3,
+					testLabels,
+					apiv1.PodRunning,
+				},
+				{
+					testPodID4,
+					testLabels,
+					apiv1.PodRunning,
+				},
+			},
+			expectedMatch: 2,
 		},
 	}
+	// Run with adding VPA first
 	for _, tc := range cases {
-		t.Run(tc.name, func(t *testing.T) {
+		t.Run(tc.name+", VPA first", func(t *testing.T) {
 			cluster := NewClusterState()
 			vpa := addVpa(cluster, testVpaID, testAnnotations, tc.vpaSelector)
 			for _, podDesc := range tc.pods {
 				cluster.AddOrUpdatePod(podDesc.id, podDesc.labels, podDesc.phase)
 				containerID := ContainerID{testPodID, "foo"}
 				assert.NoError(t, cluster.AddOrUpdateContainer(containerID, testRequest))
 			}
-			assert.Equal(t, tc.expectedMatch, cluster.VpasWithMatchingPods[vpa.ID])
+			assert.Equal(t, tc.expectedMatch, cluster.VpaPodCount[vpa.ID])
+		})
+	}
+	// Run with adding Pods first
+	for _, tc := range cases {
+		t.Run(tc.name+", Pods first", func(t *testing.T) {
+			cluster := NewClusterState()
+			for _, podDesc := range tc.pods {
+				cluster.AddOrUpdatePod(podDesc.id, podDesc.labels, podDesc.phase)
+				containerID := ContainerID{testPodID, "foo"}
+				assert.NoError(t, cluster.AddOrUpdateContainer(containerID, testRequest))
+			}
+			vpa := addVpa(cluster, testVpaID, testAnnotations, tc.vpaSelector)
+			assert.Equal(t, tc.expectedMatch, cluster.VpaPodCount[vpa.ID])
 		})
 	}
 }

vertical-pod-autoscaler/pkg/recommender/model/vpa.go

@@ -126,20 +126,18 @@ func NewVpa(id VpaID, selector labels.Selector, created time.Time) *Vpa {
 }
 
 // UseAggregationIfMatching checks if the given aggregation matches (contributes to) this VPA
-// and adds it to the set of VPA's aggregations if that is the case. Returns true
-// if the aggregation matches VPA.
-func (vpa *Vpa) UseAggregationIfMatching(aggregationKey AggregateStateKey, aggregation *AggregateContainerState) bool {
+// and adds it to the set of VPA's aggregations if that is the case.
+func (vpa *Vpa) UseAggregationIfMatching(aggregationKey AggregateStateKey, aggregation *AggregateContainerState) {
 	if vpa.UsesAggregation(aggregationKey) {
-		return true
+		// Already linked, we can return quickly.
+		return
 	}
 	if vpa.matchesAggregation(aggregationKey) {
 		vpa.aggregateContainerStates[aggregationKey] = aggregation
 		aggregation.IsUnderVPA = true
 		aggregation.UpdateMode = vpa.UpdateMode
 		aggregation.UpdateFromPolicy(vpa_api_util.GetContainerResourcePolicy(aggregationKey.ContainerName(), vpa.ResourcePolicy))
-		return true
 	}
-	return false
 }
 
 // UpdateRecommendation updates the recommended resources in the VPA and its