feat: enable pod protection based on storage classes

this commit introduces a new customization on the existing PodsWithPVC
protection. this new customization allow users to make pods that refer
to a given storage class unevictable.

for example, to protect pods referring to `storage-class-0` and
`storage-class-1` this configuration can be used:

```yaml
apiVersion: "descheduler/v1alpha2"
kind: "DeschedulerPolicy"
profiles:
- name: ProfileName
  pluginConfig:
  - name: "DefaultEvictor"
    args:
      podProtections:
        extraEnabled:
        - PodsWithPVC
        config:
          PodsWithPVC:
            protectedStorageClasses:
            - name: storage-class-0
            - name: storage-class-1
```

changes introduced by this pr:

1. the descheduler starts to observe persistent volume claims.
1. a new api field was introduced to allow per pod protection config.
1. rbac had to be adjusted (+persistentvolumeclaims).

pkg/framework/plugins/defaultevictor/defaultevictor.go

@@ -17,11 +17,13 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"maps"
 	"slices"
 
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
+	"k8s.io/client-go/informers"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/klog/v2"
 
@@ -122,13 +124,67 @@ func applyEffectivePodProtections(d *DefaultEvictor, podProtections []PodProtect
 	applyFailedBarePodsProtection(d, protectionMap)
 	applyLocalStoragePodsProtection(d, protectionMap)
 	applyDaemonSetPodsProtection(d, protectionMap)
-	applyPvcPodsProtection(d, protectionMap)
+	applyPVCPodsProtection(d, protectionMap)
 	applyPodsWithoutPDBProtection(d, protectionMap, handle)
 	applyPodsWithResourceClaimsProtection(d, protectionMap)
 
 	return nil
 }
 
+// protectedPVCStorageClasses returns the list of storage classes that should
+// be protected from eviction. If the list is empty or nil then all storage
+// classes are protected (assuming PodsWithPVC protection is enabled).
+func protectedPVCStorageClasses(d *DefaultEvictor) []ProtectedStorageClass {
+	protcfg := d.args.PodProtections.Config
+	if protcfg == nil {
+		return nil
+	}
+	scconfig := protcfg.PodsWithPVC
+	if scconfig == nil {
+		return nil
+	}
+	return scconfig.ProtectedStorageClasses
+}
+
+// podStorageClasses returns a list of storage classes referred by a pod. We
+// need this when assessing if a pod should be protected because it refers to a
+// protected storage class.
+func podStorageClasses(inf informers.SharedInformerFactory, pod *v1.Pod) ([]string, error) {
+	lister := inf.Core().V1().PersistentVolumeClaims().Lister().PersistentVolumeClaims(
+		pod.Namespace,
+	)
+
+	referred := map[string]bool{}
+	for _, vol := range pod.Spec.Volumes {
+		if vol.PersistentVolumeClaim == nil {
+			continue
+		}
+
+		claim, err := lister.Get(vol.PersistentVolumeClaim.ClaimName)
+		if err != nil {
+			return nil, fmt.Errorf(
+				"failed to get persistent volume claim %q/%q: %w",
+				pod.Namespace, vol.PersistentVolumeClaim.ClaimName, err,
+			)
+		}
+
+		// this should never happen as once a pvc is created with a nil
+		// storageClass it is automatically picked up by the default
+		// storage class. By returning an error here we make the pod
+		// protected from eviction.
+		if claim.Spec.StorageClassName == nil || *claim.Spec.StorageClassName == "" {
+			return nil, fmt.Errorf(
+				"failed to resolve storage class for pod %q/%q",
+				pod.Namespace, claim.Name,
+			)
+		}
+
+		referred[*claim.Spec.StorageClassName] = true
+	}
+
+	return slices.Collect(maps.Keys(referred)), nil
+}
+
 func applyFailedBarePodsProtection(d *DefaultEvictor, protectionMap map[PodProtection]bool) {
 	isProtectionEnabled := protectionMap[FailedBarePods]
 	if !isProtectionEnabled {
@@ -206,16 +262,50 @@ func applyDaemonSetPodsProtection(d *DefaultEvictor, protectionMap map[PodProtec
 	}
 }
 
-func applyPvcPodsProtection(d *DefaultEvictor, protectionMap map[PodProtection]bool) {
-	isProtectionEnabled := protectionMap[PodsWithPVC]
-	if isProtectionEnabled {
-		d.constraints = append(d.constraints, func(pod *v1.Pod) error {
-			if utils.IsPodWithPVC(pod) {
-				return fmt.Errorf("pod with PVC is protected against eviction")
+// applyPVCPodsProtection protects pods that refer to a PVC from eviction. If
+// the user has specified a list of storage classes to protect then only pods
+// referring to PVCs of those storage classes are protected.
+func applyPVCPodsProtection(d *DefaultEvictor, enabledProtections map[PodProtection]bool) {
+	if !enabledProtections[PodsWithPVC] {
+		return
+	}
+
+	// if the user isn't filtering by storage classes we protect all pods
+	// referring to a PVC.
+	protected := protectedPVCStorageClasses(d)
+	if len(protected) == 0 {
+		d.constraints = append(
+			d.constraints,
+			func(pod *v1.Pod) error {
+				if utils.IsPodWithPVC(pod) {
+					return fmt.Errorf("pod with PVC is protected against eviction")
+				}
+				return nil
+			},
+		)
+		return
+	}
+
+	protectedsc := map[string]bool{}
+	for _, class := range protected {
+		protectedsc[class.Name] = true
+	}
+
+	d.constraints = append(
+		d.constraints, func(pod *v1.Pod) error {
+			classes, err := podStorageClasses(d.handle.SharedInformerFactory(), pod)
+			if err != nil {
+				return err
+			}
+			for _, class := range classes {
+				if !protectedsc[class] {
+					continue
+				}
+				return fmt.Errorf("pod using protected storage class %q", class)
 			}
 			return nil
-		})
-	}
+		},
+	)
 }
 
 func applyPodsWithoutPDBProtection(d *DefaultEvictor, protectionMap map[PodProtection]bool, handle frameworktypes.Handle) {

pkg/framework/plugins/defaultevictor/defaultevictor_test.go

@@ -16,6 +16,7 @@ package defaultevictor
 import (
 	"context"
 	"fmt"
+	"reflect"
 	"slices"
 	"testing"
 	"time"
@@ -28,6 +29,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/uuid"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes/fake"
+	"k8s.io/klog/v2"
 	"k8s.io/utils/ptr"
 	"sigs.k8s.io/descheduler/pkg/api"
 	evictionutils "sigs.k8s.io/descheduler/pkg/descheduler/evictions/utils"
@@ -55,6 +57,7 @@ type testCase struct {
 	ignorePodsWithoutPDB    bool
 	podProtections          PodProtections
 	noEvictionPolicy        NoEvictionPolicy
+	pvcs                    []*v1.PersistentVolumeClaim
 }
 
 func TestDefaultEvictorPreEvictionFilter(t *testing.T) {
@@ -879,6 +882,144 @@ func TestDefaultEvictorFilter(t *testing.T) {
 			},
 			result: false,
 		},
+		{
+			description: "Pod using StorageClass is not evicted because 'PodsWithPVC' is in ExtraEnabled",
+			pods: []*v1.Pod{
+				test.BuildTestPod("p23", 400, 0, n1.Name, func(pod *v1.Pod) {
+					pod.ObjectMeta.OwnerReferences = test.GetNormalPodOwnerRefList()
+					pod.Spec.Volumes = []v1.Volume{
+						{
+							Name: "pvc", VolumeSource: v1.VolumeSource{
+								PersistentVolumeClaim: &v1.PersistentVolumeClaimVolumeSource{
+									ClaimName: "foo",
+								},
+							},
+						},
+					}
+				}),
+			},
+			podProtections: PodProtections{
+				ExtraEnabled: []PodProtection{PodsWithPVC},
+				Config: &PodProtectionsConfig{
+					PodsWithPVC: &PodsWithPVCConfig{
+						ProtectedStorageClasses: []ProtectedStorageClass{
+							{
+								Name: "standard",
+							},
+						},
+					},
+				},
+			},
+			pvcs: []*v1.PersistentVolumeClaim{
+				test.BuildTestPVC("foo", "standard"),
+			},
+			result: false,
+		},
+		{
+			description: "Pod using unprotected StorageClass is evicted even though 'PodsWithPVC' is in ExtraEnabled",
+			pods: []*v1.Pod{
+				test.BuildTestPod("p24", 400, 0, n1.Name, func(pod *v1.Pod) {
+					pod.ObjectMeta.OwnerReferences = test.GetNormalPodOwnerRefList()
+					pod.Spec.Volumes = []v1.Volume{
+						{
+							Name: "pvc", VolumeSource: v1.VolumeSource{
+								PersistentVolumeClaim: &v1.PersistentVolumeClaimVolumeSource{
+									ClaimName: "foo",
+								},
+							},
+						},
+					}
+				}),
+			},
+			podProtections: PodProtections{
+				ExtraEnabled: []PodProtection{PodsWithPVC},
+				Config: &PodProtectionsConfig{
+					PodsWithPVC: &PodsWithPVCConfig{
+						ProtectedStorageClasses: []ProtectedStorageClass{
+							{
+								Name: "protected",
+							},
+						},
+					},
+				},
+			},
+			pvcs: []*v1.PersistentVolumeClaim{
+				test.BuildTestPVC("foo", "unprotected"),
+			},
+			result: true,
+		},
+		{
+			description: "Pod using unexisting PVC is not evicted because we cannot determine if storage class is protected or not",
+			pods: []*v1.Pod{
+				test.BuildTestPod("p25", 400, 0, n1.Name, func(pod *v1.Pod) {
+					pod.ObjectMeta.OwnerReferences = test.GetNormalPodOwnerRefList()
+					pod.Spec.Volumes = []v1.Volume{
+						{
+							Name: "pvc", VolumeSource: v1.VolumeSource{
+								PersistentVolumeClaim: &v1.PersistentVolumeClaimVolumeSource{
+									ClaimName: "foo",
+								},
+							},
+						},
+					}
+				}),
+			},
+			podProtections: PodProtections{
+				ExtraEnabled: []PodProtection{PodsWithPVC},
+				Config: &PodProtectionsConfig{
+					PodsWithPVC: &PodsWithPVCConfig{
+						ProtectedStorageClasses: []ProtectedStorageClass{
+							{
+								Name: "protected",
+							},
+						},
+					},
+				},
+			},
+			pvcs:   []*v1.PersistentVolumeClaim{},
+			result: false,
+		},
+		{
+			description: "Pod using protected and unprotected StorageClasses is not evicted",
+			pods: []*v1.Pod{
+				test.BuildTestPod("p26", 400, 0, n1.Name, func(pod *v1.Pod) {
+					pod.ObjectMeta.OwnerReferences = test.GetNormalPodOwnerRefList()
+					pod.Spec.Volumes = []v1.Volume{
+						{
+							Name: "protected-pvc", VolumeSource: v1.VolumeSource{
+								PersistentVolumeClaim: &v1.PersistentVolumeClaimVolumeSource{
+									ClaimName: "protected",
+								},
+							},
+						},
+						{
+							Name: "unprotected-pvc", VolumeSource: v1.VolumeSource{
+								PersistentVolumeClaim: &v1.PersistentVolumeClaimVolumeSource{
+									ClaimName: "unprotected",
+								},
+							},
+						},
+					}
+				}),
+			},
+			podProtections: PodProtections{
+				ExtraEnabled: []PodProtection{PodsWithPVC},
+				Config: &PodProtectionsConfig{
+					PodsWithPVC: &PodsWithPVCConfig{
+						ProtectedStorageClasses: []ProtectedStorageClass{
+							{
+								Name: "protected",
+							},
+						},
+					},
+				},
+			},
+			pvcs: []*v1.PersistentVolumeClaim{
+				test.BuildTestPVC("protected", "protected"),
+				test.BuildTestPVC("unprotected", "unprotected"),
+			},
+			result: false,
+		},
 	}
 
 	for _, test := range testCases {
@@ -953,12 +1094,16 @@ func initializePlugin(ctx context.Context, test testCase) (frameworktypes.Plugin
 	for _, pdb := range test.pdbs {
 		objs = append(objs, pdb)
 	}
+	for _, pvc := range test.pvcs {
+		objs = append(objs, pvc)
+	}
 
 	fakeClient := fake.NewSimpleClientset(objs...)
 
 	sharedInformerFactory := informers.NewSharedInformerFactory(fakeClient, 0)
 	podInformer := sharedInformerFactory.Core().V1().Pods().Informer()
 	_ = sharedInformerFactory.Policy().V1().PodDisruptionBudgets().Lister()
+	_ = sharedInformerFactory.Core().V1().PersistentVolumeClaims().Lister()
 
 	getPodsAssignedToNode, err := podutil.BuildGetPodsAssignedToNodeFunc(podInformer)
 	if err != nil {
@@ -1117,3 +1262,58 @@ func slicesEqualUnordered(expected, actual []PodProtection) bool {
 	}
 	return true
 }
+
+func Test_protectedPVCStorageClasses(t *testing.T) {
+	tests := []struct {
+		name     string
+		args     *DefaultEvictorArgs
+		expected []ProtectedStorageClass
+	}{
+		{
+			name:     "no PodProtections config",
+			args:     &DefaultEvictorArgs{},
+			expected: nil,
+		},
+		{
+			name: "no PodsWithPVC config",
+			args: &DefaultEvictorArgs{
+				PodProtections: PodProtections{
+					Config: &PodProtectionsConfig{},
+				},
+			},
+			expected: nil,
+		},
+		{
+			name: "storage classes specified",
+			args: &DefaultEvictorArgs{
+				PodProtections: PodProtections{
+					Config: &PodProtectionsConfig{
+						PodsWithPVC: &PodsWithPVCConfig{
+							ProtectedStorageClasses: []ProtectedStorageClass{
+								{Name: "sc1"},
+								{Name: "sc2"},
+							},
+						},
+					},
+				},
+			},
+			expected: []ProtectedStorageClass{
+				{Name: "sc1"},
+				{Name: "sc2"},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			ev := &DefaultEvictor{
+				logger: klog.NewKlogr(),
+				args:   test.args,
+			}
+			result := protectedPVCStorageClasses(ev)
+			if !reflect.DeepEqual(result, test.expected) {
+				t.Errorf("Expected %v, got %v", test.expected, result)
+			}
+		})
+	}
+}

pkg/framework/plugins/defaultevictor/types.go

@@ -75,6 +75,37 @@ type PodProtections struct {
 	// DefaultDisabled specifies which default protection policies should be disabled.
 	// Supports: PodsWithLocalStorage, DaemonSetPods, SystemCriticalPods, FailedBarePods
 	DefaultDisabled []PodProtection `json:"defaultDisabled,omitempty"`
+
+	// Config holds configuration for pod protection policies. Depending on
+	// the enabled policies this may be required. For instance, when
+	// enabling the PodsWithPVC policy the user may specify which storage
+	// classes should be protected.
+	Config *PodProtectionsConfig `json:"config,omitempty"`
+}
+
+// PodProtectionsConfig holds configuration for pod protection policies. The
+// name of the fields here must be equal to a protection name. This struct is
+// meant to be extended as more protection policies are added.
+// +k8s:deepcopy-gen=true
+type PodProtectionsConfig struct {
+	PodsWithPVC *PodsWithPVCConfig `json:"PodsWithPVC,omitempty"`
+}
+
+// PodsWithPVCConfig holds configuration for the PodsWithPVC protection.
+// +k8s:deepcopy-gen=true
+type PodsWithPVCConfig struct {
+	// ProtectedStorageClasses is a list of storage classes that we want to
+	// protect. i.e. if a pod refers to one of these storage classes it is
+	// protected from being evicted. If none is provided then all pods with
+	// PVCs are protected from eviction.
+	ProtectedStorageClasses []ProtectedStorageClass `json:"protectedStorageClasses,omitempty"`
+}
+
+// ProtectedStorageClass is used to determine what storage classes are
+// protected when the PodsWithPVC protection is enabled. This object exists
+// so we can later on extend it with more configuration if needed.
+type ProtectedStorageClass struct {
+	Name string `json:"name"`
 }
 
 // defaultPodProtections holds the list of protection policies that are enabled by default.

pkg/framework/plugins/defaultevictor/validation.go

@@ -72,6 +72,17 @@ func ValidateDefaultEvictorArgs(obj runtime.Object) error {
 		if hasDuplicates(args.PodProtections.ExtraEnabled) {
 			allErrs = append(allErrs, fmt.Errorf("PodProtections.ExtraEnabled contains duplicate entries"))
 		}
+
+		if slices.Contains(args.PodProtections.ExtraEnabled, PodsWithPVC) {
+			if args.PodProtections.Config != nil && args.PodProtections.Config.PodsWithPVC != nil {
+				protectedsc := args.PodProtections.Config.PodsWithPVC.ProtectedStorageClasses
+				for i, sc := range protectedsc {
+					if sc.Name == "" {
+						allErrs = append(allErrs, fmt.Errorf("PodProtections.Config.PodsWithPVC.ProtectedStorageClasses[%d] name cannot be empty", i))
+					}
+				}
+			}
+		}
 	}
 
 	return utilerrors.NewAggregate(allErrs)

pkg/framework/plugins/defaultevictor/validation_test.go

@@ -198,6 +198,33 @@ func TestValidateDefaultEvictorArgs(t *testing.T) {
 			},
 			errInfo: fmt.Errorf(`[noEvictionPolicy accepts only ["Preferred" "Mandatory"] values, invalid pod protection policy in DefaultDisabled: "PodsWithoutPDB". Valid options are: [PodsWithLocalStorage SystemCriticalPods FailedBarePods DaemonSetPods], PodProtections.DefaultDisabled contains duplicate entries, PodProtections.ExtraEnabled contains duplicate entries]`),
 		},
+		{
+			name: "Protected storage classes without storage class name",
+			args: &DefaultEvictorArgs{
+				PodProtections: PodProtections{
+					ExtraEnabled: []PodProtection{PodsWithPVC},
+					Config: &PodProtectionsConfig{
+						PodsWithPVC: &PodsWithPVCConfig{
+							ProtectedStorageClasses: []ProtectedStorageClass{
+								{
+									Name: "",
+								},
+								{
+									Name: "protected-storage-class-0",
+								},
+								{
+									Name: "",
+								},
+								{
+									Name: "protected-storage-class-1",
+								},
+							},
+						},
+					},
+				},
+			},
+			errInfo: fmt.Errorf(`[PodProtections.Config.PodsWithPVC.ProtectedStorageClasses[0] name cannot be empty, PodProtections.Config.PodsWithPVC.ProtectedStorageClasses[2] name cannot be empty]`),
+		},
 	}
 
 	for _, testCase := range tests {

pkg/framework/plugins/defaultevictor/zz_generated.deepcopy.go

@@ -81,6 +81,11 @@ func (in *PodProtections) DeepCopyInto(out *PodProtections) {
 		*out = make([]PodProtection, len(*in))
 		copy(*out, *in)
 	}
+	if in.Config != nil {
+		in, out := &in.Config, &out.Config
+		*out = new(PodProtectionsConfig)
+		(*in).DeepCopyInto(*out)
+	}
 	return
 }
 
@@ -93,3 +98,45 @@ func (in *PodProtections) DeepCopy() *PodProtections {
 	in.DeepCopyInto(out)
 	return out
 }
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *PodProtectionsConfig) DeepCopyInto(out *PodProtectionsConfig) {
+	*out = *in
+	if in.PodsWithPVC != nil {
+		in, out := &in.PodsWithPVC, &out.PodsWithPVC
+		*out = new(PodsWithPVCConfig)
+		(*in).DeepCopyInto(*out)
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new PodProtectionsConfig.
+func (in *PodProtectionsConfig) DeepCopy() *PodProtectionsConfig {
+	if in == nil {
+		return nil
+	}
+	out := new(PodProtectionsConfig)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *PodsWithPVCConfig) DeepCopyInto(out *PodsWithPVCConfig) {
+	*out = *in
+	if in.ProtectedStorageClasses != nil {
+		in, out := &in.ProtectedStorageClasses, &out.ProtectedStorageClasses
+		*out = make([]ProtectedStorageClass, len(*in))
+		copy(*out, *in)
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new PodsWithPVCConfig.
+func (in *PodsWithPVCConfig) DeepCopy() *PodsWithPVCConfig {
+	if in == nil {
+		return nil
+	}
+	out := new(PodsWithPVCConfig)
+	in.DeepCopyInto(out)
+	return out
+}