fix: ensure cluster state is synced prior to scheduling

pkg/controllers/provisioning/scheduling/scheduler.go

@@ -18,10 +18,13 @@ import (
 	"context"
 	"fmt"
 	"sort"
+	"time"
 
+	"github.com/avast/retry-go"
 	"github.com/samber/lo"
 	"go.uber.org/multierr"
 	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"knative.dev/pkg/logging"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
@@ -33,6 +36,10 @@ import (
 	"github.com/aws/karpenter/pkg/utils/resources"
 )
 
+// ClusterSyncRetries controls how many times we attempt to retry waiting on cluster state sync. This is exposed for
+// unit testing purposes so we can avoid a lengthy delay in cluster sync.
+var ClusterSyncRetries uint = 5
+
 func NewScheduler(ctx context.Context, kubeClient client.Client, nodeTemplates []*scheduling.NodeTemplate, provisioners []v1alpha5.Provisioner, cluster *state.Cluster, topology *Topology, instanceTypes map[string][]cloudprovider.InstanceType, daemonOverhead map[*scheduling.NodeTemplate]v1.ResourceList, recorder events.Recorder) *Scheduler {
 	for provisioner := range instanceTypes {
 		sort.Slice(instanceTypes[provisioner], func(i, j int) bool {
@@ -62,6 +69,9 @@ func NewScheduler(ctx context.Context, kubeClient client.Client, nodeTemplates [
 		}
 	}
 
+	// wait to ensure that our cluster state is synced with the current known nodes to prevent over-shooting
+	s.waitForClusterStateSync(ctx)
+
 	// create our in-flight nodes
 	s.cluster.ForEachNode(func(node *state.Node) bool {
 		name, ok := node.Node.Labels[v1alpha5.ProvisionerNameLabelKey]
@@ -224,6 +234,40 @@ func (s *Scheduler) add(ctx context.Context, pod *v1.Pod) error {
 	return errs
 }
 
+// waitForClusterStateSync ensures that our cluster state is aware of at least all of the nodes that our list cache has.
+// Since we launch nodes in parallel, we can create many node objects which may not all be reconciled by the cluster
+// state before we start trying to schedule again.  In this case, we would over-provision as we weren't aware of the
+// inflight nodes.
+func (s *Scheduler) waitForClusterStateSync(ctx context.Context) {
+	if err := retry.Do(func() error {
+		// collect the nodes known by the kube API server
+		var nodes v1.NodeList
+		if err := s.kubeClient.List(ctx, &nodes); err != nil {
+			return nil
+		}
+		unknownNodes := sets.NewString()
+		for _, n := range nodes.Items {
+			unknownNodes.Insert(n.Name)
+		}
+
+		// delete any that cluster state already knows about
+		s.cluster.ForEachNode(func(n *state.Node) bool {
+			delete(unknownNodes, n.Node.Name)
+			return true
+		})
+
+		// and we're left with nodes which exist, but haven't reconciled with cluster state yet
+		if len(unknownNodes) != 0 {
+			return fmt.Errorf("%d nodes not known to cluster state", len(unknownNodes))
+		}
+		return nil
+	}, retry.Delay(1*time.Second),
+		retry.Attempts(ClusterSyncRetries),
+	); err != nil {
+		logging.FromContext(ctx).Infof("nodes failed to sync, may launch too many nodes which should resolve")
+	}
+}
+
 // subtractMax returns the remaining resources after subtracting the max resource quantity per instance type. To avoid
 // overshooting out, we need to pessimistically assume that if e.g. we request a 2, 4 or 8 CPU instance type
 // that the 8 CPU instance type is all that will be available.  This could cause a batch of pods to take multiple rounds

pkg/controllers/provisioning/scheduling/suite_test.go

@@ -66,6 +66,7 @@ func TestAPIs(t *testing.T) {
 }
 
 var _ = BeforeSuite(func() {
+	scheduling.ClusterSyncRetries = 0
 	env = test.NewEnvironment(ctx, func(e *test.Environment) {
 		cloudProv = &fake.CloudProvider{}
 		cfg = test.NewConfig()

pkg/controllers/state/cluster.go

@@ -292,9 +292,11 @@ func (c *Cluster) deleteNode(nodeName string) {
 
 // updateNode is called for every node reconciliation
 func (c *Cluster) updateNode(ctx context.Context, node *v1.Node) error {
+	// perform node lookup before we lock so that the slower operation can occur in parallel
+	n, err := c.newNode(ctx, node)
+
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	n, err := c.newNode(ctx, node)
 	if err != nil {
 		// ensure that the out of date node is forgotten
 		delete(c.nodes, node.Name)

pkg/controllers/state/node.go

@@ -22,6 +22,7 @@ import (
 	"knative.dev/pkg/logging"
 	controllerruntime "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )
@@ -65,5 +66,6 @@ func (c *NodeController) Register(ctx context.Context, m manager.Manager) error
 		NewControllerManagedBy(m).
 		Named(nodeControllerName).
 		For(&v1.Node{}).
+		WithOptions(controller.Options{MaxConcurrentReconciles: 10}).
 		Complete(c)
 }

pkg/controllers/state/pod.go

@@ -23,6 +23,7 @@ import (
 	"knative.dev/pkg/logging"
 	controllerruntime "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )
@@ -66,6 +67,7 @@ func (c *PodController) Register(ctx context.Context, m manager.Manager) error {
 	return controllerruntime.
 		NewControllerManagedBy(m).
 		Named(podControllerName).
+		WithOptions(controller.Options{MaxConcurrentReconciles: 10}).
 		For(&v1.Pod{}).
 		Complete(c)
 }