From df9dab258a9fc0a76d96952d8a67df0c3b59a48c Mon Sep 17 00:00:00 2001
From: Todd Neal <tnealt@amazon.com>
Date: Wed, 18 Jan 2023 13:09:35 -0600
Subject: [PATCH] fix: validate volume names (#165)

We need to validate volume names if specified to ensure they exist so we
don't launch a node for pods that won't schedule anyway.

Fixes aws/karpenter/#3219
---
 pkg/controllers/provisioning/suite_test.go    | 11 ++++++
 .../provisioning/volumetopology.go            | 37 +++++++++++++++----
 2 files changed, 41 insertions(+), 7 deletions(-)

diff --git a/pkg/controllers/provisioning/suite_test.go b/pkg/controllers/provisioning/suite_test.go
index 8048a94488..35d52739f0 100644
--- a/pkg/controllers/provisioning/suite_test.go
+++ b/pkg/controllers/provisioning/suite_test.go
@@ -940,6 +940,17 @@ var _ = Describe("Volume Topology Requirements", func() {
 		ExpectNotScheduled(ctx, env.Client, invalidPod)
 		ExpectScheduled(ctx, env.Client, pod)
 	})
+	It("should schedule valid pods when a pod with an invalid pvc is encountered (volume name)", func() {
+		invalidVolumeName := "invalid-volume-name"
+		persistentVolumeClaim := test.PersistentVolumeClaim(test.PersistentVolumeClaimOptions{VolumeName: invalidVolumeName})
+		ExpectApplied(ctx, env.Client, test.Provisioner(), persistentVolumeClaim)
+		invalidPod := ExpectProvisioned(ctx, env.Client, cluster, recorder, provisioningController, prov, test.UnschedulablePod(test.PodOptions{
+			PersistentVolumeClaims: []string{persistentVolumeClaim.Name},
+		}))[0]
+		pod := ExpectProvisioned(ctx, env.Client, cluster, recorder, provisioningController, prov, test.UnschedulablePod(test.PodOptions{}))[0]
+		ExpectNotScheduled(ctx, env.Client, invalidPod)
+		ExpectScheduled(ctx, env.Client, pod)
+	})
 	It("should schedule to storage class zones if volume does not exist", func() {
 		persistentVolumeClaim := test.PersistentVolumeClaim(test.PersistentVolumeClaimOptions{StorageClassName: &storageClass.Name})
 		ExpectApplied(ctx, env.Client, test.Provisioner(), storageClass, persistentVolumeClaim)
diff --git a/pkg/controllers/provisioning/volumetopology.go b/pkg/controllers/provisioning/volumetopology.go
index 54e583991b..097d8a42d1 100644
--- a/pkg/controllers/provisioning/volumetopology.go
+++ b/pkg/controllers/provisioning/volumetopology.go
@@ -145,6 +145,7 @@ func (v *VolumeTopology) getPersistentVolumeClaim(ctx context.Context, pod *v1.P
 func (v *VolumeTopology) validatePersistentVolumeClaims(ctx context.Context, pod *v1.Pod) error {
 	for _, volume := range pod.Spec.Volumes {
 		var storageClassName *string
+		var volumeName string
 		if volume.PersistentVolumeClaim != nil {
 			// validate the PVC if it exists
 			pvc, err := v.getPersistentVolumeClaim(ctx, pod, volume)
@@ -154,19 +155,41 @@ func (v *VolumeTopology) validatePersistentVolumeClaims(ctx context.Context, pod
 			// may not have a PVC
 			if pvc == nil {
 				continue
-
 			}
+
 			storageClassName = pvc.Spec.StorageClassName
+			volumeName = pvc.Spec.VolumeName
 		} else if volume.Ephemeral != nil {
 			storageClassName = volume.Ephemeral.VolumeClaimTemplate.Spec.StorageClassName
 		}
 
-		// we have a storage class name, so ensure that it exists
-		if storageClassName != nil && ptr.StringValue(storageClassName) != "" {
-			storageClass := &storagev1.StorageClass{}
-			if err := v.kubeClient.Get(ctx, types.NamespacedName{Name: ptr.StringValue(storageClassName)}, storageClass); err != nil {
-				return err
-			}
+		if err := v.validateStorageClass(ctx, storageClassName); err != nil {
+			return err
+		}
+		if err := v.validateVolume(ctx, volumeName); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (v *VolumeTopology) validateVolume(ctx context.Context, volumeName string) error {
+	// we have a volume name, so ensure that it exists
+	if volumeName != "" {
+		pv := &v1.PersistentVolume{}
+		if err := v.kubeClient.Get(ctx, types.NamespacedName{Name: volumeName}, pv); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (v *VolumeTopology) validateStorageClass(ctx context.Context, storageClassName *string) error {
+	// we have a storage class name, so ensure that it exists
+	if ptr.StringValue(storageClassName) != "" {
+		storageClass := &storagev1.StorageClass{}
+		if err := v.kubeClient.Get(ctx, types.NamespacedName{Name: ptr.StringValue(storageClassName)}, storageClass); err != nil {
+			return err
 		}
 	}
 	return nil