fix: track volume mounts per node

Assume nodes can support infinite volumes until they launch. Once the CSI driver is
reporting the current number of mountable volumes, use that value which may require
launching more nodes. Remove the PVC controller since we no longer bind pods to
nodes.

Fixes #919 and #1888

charts/karpenter/templates/clusterrole.yaml

@@ -39,3 +39,6 @@ rules:
   - apiGroups: ["admissionregistration.k8s.io"]
     resources: ["validatingwebhookconfigurations", "mutatingwebhookconfigurations"]
     verbs: ["get", "watch", "list", "update"]
+  - apiGroups: ["storage.k8s.io"]
+    resources: ["csinodes"]
+    verbs: ["get", "watch", "list"]

cmd/controller/main.go

@@ -47,7 +47,6 @@ import (
 	metricspod "github.com/aws/karpenter/pkg/controllers/metrics/pod"
 	metricsprovisioner "github.com/aws/karpenter/pkg/controllers/metrics/provisioner"
 	"github.com/aws/karpenter/pkg/controllers/node"
-	"github.com/aws/karpenter/pkg/controllers/persistentvolumeclaim"
 	"github.com/aws/karpenter/pkg/controllers/provisioning"
 	"github.com/aws/karpenter/pkg/controllers/state"
 	"github.com/aws/karpenter/pkg/controllers/termination"
@@ -110,13 +109,12 @@ func main() {
 		logging.FromContext(ctx).Errorf("watching configmaps, config changes won't be applied immediately, %s", err)
 	}
 
-	cluster := state.NewCluster(ctx, manager.GetClient(), cloudProvider)
+	cluster := state.NewCluster(manager.GetClient(), cloudProvider)
 
 	if err := manager.RegisterControllers(ctx,
 		provisioning.NewController(ctx, cfg, manager.GetClient(), clientSet.CoreV1(), recorder, cloudProvider, cluster),
 		state.NewNodeController(manager.GetClient(), cluster),
 		state.NewPodController(manager.GetClient(), cluster),
-		persistentvolumeclaim.NewController(manager.GetClient()),
 		termination.NewController(ctx, manager.GetClient(), clientSet.CoreV1(), cloudProvider),
 		node.NewController(manager.GetClient(), cloudProvider),
 		metricspod.NewController(manager.GetClient()),

hack/docs/metrics_gen_docs.go

@@ -205,6 +205,12 @@ func handleVariableDeclaration(v *ast.GenDecl) []metricInfo {
 }
 
 func getFuncPackage(fun ast.Expr) string {
+	if pexpr, ok := fun.(*ast.ParenExpr); ok {
+		return getFuncPackage(pexpr.X)
+	}
+	if sexpr, ok := fun.(*ast.StarExpr); ok {
+		return getFuncPackage(sexpr.X)
+	}
 	if sel, ok := fun.(*ast.SelectorExpr); ok {
 		return fmt.Sprintf("%s", sel.X)
 	}

pkg/cloudprovider/aws/instancetype.go

@@ -38,6 +38,8 @@ import (
 	"github.com/aws/karpenter/pkg/utils/sets"
 )
 
+var _ cloudprovider.InstanceType = (*InstanceType)(nil)
+
 type InstanceType struct {
 	*ec2.InstanceTypeInfo
 	offerings    []cloudprovider.Offering

pkg/cloudprovider/aws/suite_test.go

@@ -130,7 +130,7 @@ var _ = BeforeSuite(func() {
 			kubeClient: e.Client,
 		}
 		registry.RegisterOrDie(ctx, cloudProvider)
-		cluster = state.NewCluster(ctx, e.Client, cloudProvider)
+		cluster = state.NewCluster(e.Client, cloudProvider)
 		recorder = test.NewEventRecorder()
 		cfg = test.NewConfig()
 		controller = provisioning.NewController(ctx, cfg, e.Client, clientSet.CoreV1(), recorder, cloudProvider, cluster)

pkg/cloudprovider/fake/cloudprovider.go

@@ -49,6 +49,9 @@ type CloudProvider struct {
 	CreateCalls []*cloudprovider.NodeRequest
 }
 
+var _ cloudprovider.CloudProvider = (*CloudProvider)(nil)
+var _ cloudprovider.InstanceType = (*InstanceType)(nil)
+
 func (c *CloudProvider) Create(ctx context.Context, nodeRequest *cloudprovider.NodeRequest) (*v1.Node, error) {
 	c.mu.Lock()
 	c.CreateCalls = append(c.CreateCalls, nodeRequest)

pkg/config/suite_test.go

@@ -124,13 +124,7 @@ var _ = Describe("Batch Parameter", func() {
 		var changed int64
 		cfg.OnChange(func(c config.Config) {
 			defer GinkgoRecover()
-			// we can't unregister this, so just check for the one case we care about
-			if atomic.LoadInt64(&changed) == 0 {
-				atomic.StoreInt64(&changed, 1)
-				Expect(cfg.BatchIdleDuration()).To(Equal(1 * time.Second))
-				// shouldn't be changed
-				Expect(cfg.BatchMaxDuration()).To(Equal(10 * time.Second))
-			}
+			atomic.StoreInt64(&changed, 1)
 		})
 
 		// simulate user updating the config map with a bad max duration

pkg/controllers/provisioning/provisioner.go

@@ -224,7 +224,7 @@ func (p *Provisioner) schedule(ctx context.Context, pods []*v1.Pod) ([]*schedule
 		return nil, fmt.Errorf("getting daemon overhead, %w", err)
 	}
 
-	return scheduler.NewScheduler(nodeTemplates, provisionerList.Items, p.cluster, topology, instanceTypes, daemonOverhead, p.recorder).Solve(ctx, pods)
+	return scheduler.NewScheduler(ctx, p.kubeClient, nodeTemplates, provisionerList.Items, p.cluster, topology, instanceTypes, daemonOverhead, p.recorder).Solve(ctx, pods)
 }
 
 func (p *Provisioner) launch(ctx context.Context, node *scheduler.Node) error {

pkg/controllers/provisioning/scheduling/inflightnode.go

@@ -15,6 +15,7 @@ limitations under the License.
 package scheduling
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/aws/karpenter/pkg/apis/provisioning/v1alpha5"
@@ -35,20 +36,23 @@ type InFlightNode struct {
 	requirements       scheduling.Requirements
 	available          v1.ResourceList
 	startupTolerations []v1.Toleration
-	hostPortUsage      *state.HostPortUsage
+	hostPortUsage      *scheduling.HostPortUsage
+	volumeUsage        *scheduling.VolumeLimits
+	volumeLimits       scheduling.VolumeCount
 }
 
 func NewInFlightNode(n *state.Node, topology *Topology, startupTaints []v1.Taint, daemonResources v1.ResourceList) *InFlightNode {
 	// the remaining daemonResources to schedule are the total daemonResources minus what has already scheduled
 	remainingDaemonResources := resources.Subtract(daemonResources, n.DaemonSetRequested)
-
 	node := &InFlightNode{
 		Node:          n.Node,
 		available:     n.Available,
 		topology:      topology,
 		requests:      remainingDaemonResources,
 		requirements:  scheduling.NewLabelRequirements(n.Node.Labels),
 		hostPortUsage: n.HostPortUsage.Copy(),
+		volumeUsage:   n.VolumeUsage.Copy(),
+		volumeLimits:  n.VolumeLimits,
 	}
 
 	if n.Node.Labels[v1alpha5.LabelNodeInitialized] != "true" {
@@ -81,15 +85,24 @@ func NewInFlightNode(n *state.Node, topology *Topology, startupTaints []v1.Taint
 	return node
 }
 
-func (n *InFlightNode) Add(pod *v1.Pod) error {
+func (n *InFlightNode) Add(ctx context.Context, pod *v1.Pod) error {
 	// Check Taints
 	if err := scheduling.Taints(n.Node.Spec.Taints).Tolerates(pod, n.startupTolerations...); err != nil {
 		return err
 	}
 
-	if err := n.hostPortUsage.Add(pod); err != nil {
+	if err := n.hostPortUsage.Validate(pod); err != nil {
+		return err
+	}
+
+	// determine the number of volumes that will be mounted if the pod schedules
+	mountedVolumeCount, err := n.volumeUsage.Validate(ctx, pod)
+	if err != nil {
 		return err
 	}
+	if mountedVolumeCount.Exceeds(n.volumeLimits) {
+		return fmt.Errorf("would exceed node volume limits")
+	}
 
 	// check resource requests first since that's a pretty likely reason the pod won't schedule on an in-flight
 	// node, which at this point can't be increased in size
@@ -122,5 +135,7 @@ func (n *InFlightNode) Add(pod *v1.Pod) error {
 	n.requests = requests
 	n.requirements = nodeRequirements
 	n.topology.Record(pod, nodeRequirements)
+	n.hostPortUsage.Add(ctx, pod)
+	n.volumeUsage.Add(ctx, pod)
 	return nil
 }