Cluster-autoscaler: decrease node group target size if no new nodes arrive in reasonable time

cluster-autoscaler/cluster_autoscaler.go

@@ -303,6 +303,19 @@ func run(_ <-chan struct{}) {
 					}
 				}
 
+				// Check if there has been a constant difference between the number of nodes in k8s and
+				// the number of nodes on the cloud provider side.
+				// TODO: andrewskim - add protection for ready AWS nodes.
+				fixedSomething, err := fixNodeGroupSize(&autoscalingContext, time.Now())
+				if err != nil {
+					glog.Warningf("Failed to fix node group sizes: %v", err)
+					continue
+				}
+				if fixedSomething {
+					glog.V(0).Infof("Some node group target size was fixed, skipping the iteration")
+					continue
+				}
+
 				// TODO: remove once all of the unready node handling elements are in place.
 				if err := CheckGroupsAndNodes(readyNodes, autoscalingContext.CloudProvider); err != nil {
 					glog.Warningf("Cluster is not ready for autoscaling: %v", err)

cluster-autoscaler/utils.go

@@ -244,3 +244,31 @@ func removeOldUnregisteredNodes(unregisteredNodes []clusterstate.UnregisteredNod
 	}
 	return removedAny, nil
 }
+
+// Sets the target size of node groups to the current number of nodes in them
+// if the difference was constant for a prolonged time. Returns true if managed
+// to fix something.
+func fixNodeGroupSize(contetxt *AutoscalingContext, currentTime time.Time) (bool, error) {
+	fixed := false
+	for _, nodeGroup := range contetxt.CloudProvider.NodeGroups() {
+		incorrectSize := contetxt.ClusterStateRegistry.GetIncorrectNodeGroupSize(nodeGroup.Id())
+		if incorrectSize == nil {
+			continue
+		}
+		if incorrectSize.FirstObserved.Add(contetxt.UnregisteredNodeRemovalTime).Before(currentTime) {
+
+			delta := incorrectSize.CurrentSize - incorrectSize.ExpectedSize
+			if delta < 0 {
+				glog.V(0).Infof("Decreasing size of %s, expected=%d current=%d delta=%d", nodeGroup.Id(),
+					incorrectSize.ExpectedSize,
+					incorrectSize.CurrentSize,
+					delta)
+				if err := nodeGroup.DecreaseTargetSize(delta); err != nil {
+					return fixed, fmt.Errorf("Failed to decrease %s: %v", nodeGroup.Id(), err)
+				}
+				fixed = true
+			}
+		}
+	}
+	return fixed, nil
+}

cluster-autoscaler/utils_test.go

@@ -100,3 +100,42 @@ func TestRemoveOldUnregisteredNodes(t *testing.T) {
 	deletedNode := getStringFromChan(deletedNodes)
 	assert.Equal(t, "ng1/ng1-2", deletedNode)
 }
+
+func TestRemoveFixNodeTargetSize(t *testing.T) {
+	sizeChanges := make(chan string, 10)
+	now := time.Now()
+
+	ng1_1 := BuildTestNode("ng1-1", 1000, 1000)
+	ng1_1.Spec.ProviderID = "ng1-1"
+	provider := testprovider.NewTestCloudProvider(func(nodegroup string, delta int) error {
+		sizeChanges <- fmt.Sprintf("%s/%d", nodegroup, delta)
+		return nil
+	}, nil)
+	provider.AddNodeGroup("ng1", 1, 10, 3)
+	provider.AddNode("ng1", ng1_1)
+
+	clusterState := clusterstate.NewClusterStateRegistry(provider, clusterstate.ClusterStateRegistryConfig{
+		MaxTotalUnreadyPercentage: 10,
+		OkTotalUnreadyCount:       1,
+	})
+	err := clusterState.UpdateNodes([]*apiv1.Node{ng1_1}, now.Add(-time.Hour))
+	assert.NoError(t, err)
+
+	context := &AutoscalingContext{
+		CloudProvider:               provider,
+		ClusterStateRegistry:        clusterState,
+		UnregisteredNodeRemovalTime: 45 * time.Minute,
+	}
+
+	// Nothing should be fixed. The incorrect size state is not old enough.
+	removed, err := fixNodeGroupSize(context, now.Add(-50*time.Minute))
+	assert.NoError(t, err)
+	assert.False(t, removed)
+
+	// Node group should be decreased.
+	removed, err = fixNodeGroupSize(context, now)
+	assert.NoError(t, err)
+	assert.True(t, removed)
+	change := getStringFromChan(sizeChanges)
+	assert.Equal(t, "ng1/-2", change)
+}