Merge pull request #125097 from YamasouA/ft/queuehit-csinode

volumebinding: scheduler queueing hints - CSINode

pkg/scheduler/framework/plugins/volumebinding/volume_binding.go

@@ -24,6 +24,7 @@ import (
 	"time"
 
 	v1 "k8s.io/api/core/v1"
+	storagev1 "k8s.io/api/storage/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
 	corelisters "k8s.io/client-go/listers/core/v1"
@@ -35,6 +36,7 @@ import (
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/feature"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/helper"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/names"
+	"k8s.io/kubernetes/pkg/scheduler/util"
 )
 
 const (
@@ -114,7 +116,8 @@ func (pl *VolumeBinding) EventsToRegister() []framework.ClusterEventWithHint {
 		// See: https://github.com/kubernetes/kubernetes/issues/110175
 		{Event: framework.ClusterEvent{Resource: framework.Node, ActionType: framework.Add | framework.UpdateNodeLabel | framework.UpdateNodeTaint}},
 		// We rely on CSI node to translate in-tree PV to CSI.
-		{Event: framework.ClusterEvent{Resource: framework.CSINode, ActionType: framework.Add | framework.Update}},
+		// TODO: kube-schduler will unregister the CSINode events once all the volume plugins has completed their CSI migration.
+		{Event: framework.ClusterEvent{Resource: framework.CSINode, ActionType: framework.Add | framework.Update}, QueueingHintFn: pl.isSchedulableAfterCSINodeChange},
 		// When CSIStorageCapacity is enabled, pods may become schedulable
 		// on CSI driver & storage capacity changes.
 		{Event: framework.ClusterEvent{Resource: framework.CSIDriver, ActionType: framework.Add | framework.Update}},
@@ -123,6 +126,31 @@ func (pl *VolumeBinding) EventsToRegister() []framework.ClusterEventWithHint {
 	return events
 }
 
+func (pl *VolumeBinding) isSchedulableAfterCSINodeChange(logger klog.Logger, pod *v1.Pod, oldObj, newObj interface{}) (framework.QueueingHint, error) {
+	if oldObj == nil {
+		logger.V(5).Info("CSINode creation could make the pod schedulable")
+		return framework.Queue, nil
+	}
+	oldCSINode, modifiedCSINode, err := util.As[*storagev1.CSINode](oldObj, newObj)
+	if err != nil {
+		return framework.Queue, err
+	}
+
+	logger = klog.LoggerWithValues(
+		logger,
+		"Pod", klog.KObj(pod),
+		"CSINode", klog.KObj(modifiedCSINode),
+	)
+
+	if oldCSINode.ObjectMeta.Annotations[v1.MigratedPluginsAnnotationKey] != modifiedCSINode.ObjectMeta.Annotations[v1.MigratedPluginsAnnotationKey] {
+		logger.V(5).Info("CSINode's migrated plugins annotation is updated and that may make the pod schedulable")
+		return framework.Queue, nil
+	}
+
+	logger.V(5).Info("CISNode was created or updated but it doesn't make this pod schedulable")
+	return framework.QueueSkip, nil
+}
+
 // podHasPVCs returns 2 values:
 // - the first one to denote if the given "pod" has any PVC defined.
 // - the second one to return any error if the requested PVC is illegal.

pkg/scheduler/framework/plugins/volumebinding/volume_binding_test.go

@@ -890,3 +890,109 @@ func TestVolumeBinding(t *testing.T) {
 		})
 	}
 }
+
+func TestIsSchedulableAfterCSINodeChange(t *testing.T) {
+	table := []struct {
+		name   string
+		oldObj interface{}
+		newObj interface{}
+		err    bool
+		expect framework.QueueingHint
+	}{
+		{
+			name:   "unexpected objects are passed",
+			oldObj: new(struct{}),
+			newObj: new(struct{}),
+			err:    true,
+			expect: framework.Queue,
+		},
+		{
+			name: "CSINode is newly created",
+			newObj: &storagev1.CSINode{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "csinode-a",
+				},
+			},
+			oldObj: nil,
+			err:    false,
+			expect: framework.Queue,
+		},
+		{
+			name: "CSINode's migrated-plugins annotations is added",
+			oldObj: &storagev1.CSINode{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "csinode-a",
+					Annotations: map[string]string{
+						v1.MigratedPluginsAnnotationKey: "test1",
+					},
+				},
+			},
+			newObj: &storagev1.CSINode{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "csinode-a",
+					Annotations: map[string]string{
+						v1.MigratedPluginsAnnotationKey: "test1, test2",
+					},
+				},
+			},
+			err:    false,
+			expect: framework.Queue,
+		},
+		{
+			name: "CSINode's migrated-plugins annotation is updated",
+			oldObj: &storagev1.CSINode{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "csinode-a",
+					Annotations: map[string]string{
+						v1.MigratedPluginsAnnotationKey: "test1",
+					},
+				},
+			},
+			newObj: &storagev1.CSINode{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "csinode-a",
+					Annotations: map[string]string{
+						v1.MigratedPluginsAnnotationKey: "test2",
+					},
+				},
+			},
+			err:    false,
+			expect: framework.Queue,
+		},
+		{
+			name: "CSINode is updated but migrated-plugins annotation gets unchanged",
+			oldObj: &storagev1.CSINode{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "csinode-a",
+					Annotations: map[string]string{
+						v1.MigratedPluginsAnnotationKey: "test1",
+					},
+				},
+			},
+			newObj: &storagev1.CSINode{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "csinode-a",
+					Annotations: map[string]string{
+						v1.MigratedPluginsAnnotationKey: "test1",
+					},
+				},
+			},
+			err:    false,
+			expect: framework.QueueSkip,
+		},
+	}
+	for _, item := range table {
+		t.Run(item.name, func(t *testing.T) {
+			pl := &VolumeBinding{}
+			pod := makePod("pod-a").Pod
+			logger, _ := ktesting.NewTestContext(t)
+			qhint, err := pl.isSchedulableAfterCSINodeChange(logger, pod, item.oldObj, item.newObj)
+			if (err != nil) != item.err {
+				t.Errorf("isSchedulableAfterCSINodeChange failed - got: %q", err)
+			}
+			if qhint != item.expect {
+				t.Errorf("QHint does not match: %v, want: %v", qhint, item.expect)
+			}
+		})
+	}
+}