modify: support rollback

go.mod

@@ -18,6 +18,7 @@ require (
 	k8s.io/component-base v0.33.1
 	k8s.io/csi-translation-lib v0.33.0
 	k8s.io/klog/v2 v2.130.1
+	k8s.io/utils v0.0.0-20241210054802-24370beab758
 )
 
 require (
@@ -69,7 +70,6 @@ require (
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	k8s.io/kube-openapi v0.0.0-20250318190949-c8a335a9a2ff // indirect
-	k8s.io/utils v0.0.0-20241210054802-24370beab758 // indirect
 	sigs.k8s.io/json v0.0.0-20241014173422-cfa47c3a1cc8 // indirect
 	sigs.k8s.io/randfill v1.0.0 // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.6.0 // indirect

pkg/controller/controller.go

@@ -291,7 +291,7 @@ func (ctrl *resizeController) Run(workers int, ctx context.Context) {
 		go ctrl.slowSet.Run(stopCh)
 	}
 
-	for i := 0; i < workers; i++ {
+	for range workers {
 		go wait.Until(ctrl.syncPVCs, 0, stopCh)
 	}
 

pkg/controller/controller_test.go

@@ -211,7 +211,7 @@ func TestController(t *testing.T) {
 			disableVolumeInUseErrorHandler: true,
 		},
 	} {
-		client := csi.NewMockClient("mock", test.NodeResize, true, false, true, true, false)
+		client := csi.NewMockClient("mock", test.NodeResize, true, false, true, true)
 		driverName, _ := client.GetDriverName(context.TODO())
 
 		var expectedCap resource.Quantity
@@ -378,7 +378,7 @@ func TestResizePVC(t *testing.T) {
 		},
 	} {
 		t.Run(test.Name, func(t *testing.T) {
-			client := csi.NewMockClient("mock", test.NodeResize, true, false, true, true, false)
+			client := csi.NewMockClient("mock", test.NodeResize, true, false, true, true)
 			if test.expansionError != nil {
 				client.SetExpansionError(test.expansionError)
 			}

pkg/controller/expand_and_recover_test.go

@@ -159,7 +159,7 @@ func TestExpandAndRecover(t *testing.T) {
 		test := tests[i]
 		t.Run(test.name, func(t *testing.T) {
 			featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.RecoverVolumeExpansionFailure, true)
-			client := csi.NewMockClient("foo", !test.disableNodeExpansion, !test.disableControllerExpansion, false, true, true, false)
+			client := csi.NewMockClient("foo", !test.disableNodeExpansion, !test.disableControllerExpansion, false, true, true)
 			driverName, _ := client.GetDriverName(context.TODO())
 			if test.expansionError != nil {
 				client.SetExpansionError(test.expansionError)

pkg/controller/resize_status_test.go

@@ -77,7 +77,7 @@ func TestResizeFunctions(t *testing.T) {
 		tc := test
 		t.Run(tc.name, func(t *testing.T) {
 			featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.RecoverVolumeExpansionFailure, true)
-			client := csi.NewMockClient("foo", true, true, false, true, true, false)
+			client := csi.NewMockClient("foo", true, true, false, true, true)
 			driverName, _ := client.GetDriverName(context.TODO())
 
 			pvc := test.pvc

pkg/csi/mock_client.go

@@ -3,6 +3,8 @@ package csi
 import (
 	"context"
 	"fmt"
+	"maps"
+	"sync"
 	"sync/atomic"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
@@ -16,7 +18,6 @@ func NewMockClient(
 	supportsControllerModify bool,
 	supportsPluginControllerService bool,
 	supportsControllerSingleNodeMultiWriter bool,
-	supportsExtraModifyMetada bool,
 ) *MockClient {
 	return &MockClient{
 		name:                                    name,
@@ -25,7 +26,7 @@ func NewMockClient(
 		supportsControllerModify:                supportsControllerModify,
 		supportsPluginControllerService:         supportsPluginControllerService,
 		supportsControllerSingleNodeMultiWriter: supportsControllerSingleNodeMultiWriter,
-		extraModifyMetadata:                     supportsExtraModifyMetada,
+		modifiedParameters:                      make(map[string]string),
 	}
 }
 
@@ -43,7 +44,8 @@ type MockClient struct {
 	checkMigratedLabel                      bool
 	usedSecrets                             atomic.Pointer[map[string]string]
 	usedCapability                          atomic.Pointer[csi.VolumeCapability]
-	extraModifyMetadata                     bool
+	modifyMu                                sync.Mutex
+	modifiedParameters                      map[string]string
 }
 
 func (c *MockClient) GetDriverName(context.Context) (string, error) {
@@ -116,6 +118,12 @@ func (c *MockClient) GetModifyCount() int {
 	return int(c.modifyCalled.Load())
 }
 
+func (c *MockClient) GetModifiedParameters() map[string]string {
+	c.modifyMu.Lock()
+	defer c.modifyMu.Unlock()
+	return maps.Clone(c.modifiedParameters)
+}
+
 func (c *MockClient) GetCapability() *csi.VolumeCapability {
 	return c.usedCapability.Load()
 }
@@ -138,5 +146,8 @@ func (c *MockClient) Modify(
 	if c.modifyError != nil {
 		return c.modifyError
 	}
+	c.modifyMu.Lock()
+	defer c.modifyMu.Unlock()
+	maps.Copy(c.modifiedParameters, mutableParameters)
 	return nil
 }

pkg/modifier/csi_modifier_test.go

@@ -28,7 +28,7 @@ func TestNewModifier(t *testing.T) {
 			SupportsControllerModify: false,
 		},
 	} {
-		client := csi.NewMockClient("mock", false, false, c.SupportsControllerModify, false, false, false)
+		client := csi.NewMockClient("mock", false, false, c.SupportsControllerModify, false, false)
 		driverName := "mock-driver"
 		k8sClient, informerFactory := fakeK8s()
 		_, err := NewModifierFromClient(client, 0, k8sClient, informerFactory, false, driverName)

pkg/modifycontroller/controller.go

@@ -19,6 +19,7 @@ package modifycontroller
 import (
 	"context"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/kubernetes-csi/external-resizer/pkg/util"
@@ -38,6 +39,7 @@ import (
 	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog/v2"
+	"k8s.io/utils/ptr"
 )
 
 // ModifyController watches PVCs and checks if they are requesting an modify operation.
@@ -61,7 +63,7 @@ type modifyController struct {
 	vacListerSynced     cache.InformerSynced
 	extraModifyMetadata bool
 	// the key of the map is {PVC_NAMESPACE}/{PVC_NAME}
-	uncertainPVCs map[string]v1.PersistentVolumeClaim
+	uncertainPVCs sync.Map
 	// slowSet tracks PVCs for which modification failed with infeasible error and should be retried at slower rate.
 	slowSet *slowset.SlowSet
 }
@@ -121,19 +123,18 @@ func NewModifyController(
 }
 
 func (ctrl *modifyController) initUncertainPVCs() error {
-	ctrl.uncertainPVCs = make(map[string]v1.PersistentVolumeClaim)
 	allPVCs, err := ctrl.pvcLister.List(labels.Everything())
 	if err != nil {
 		klog.Errorf("Failed to list pvcs when init uncertain pvcs: %v", err)
 		return err
 	}
 	for _, pvc := range allPVCs {
-		if pvc.Status.ModifyVolumeStatus != nil && (pvc.Status.ModifyVolumeStatus.Status == v1.PersistentVolumeClaimModifyVolumeInProgress || pvc.Status.ModifyVolumeStatus.Status == v1.PersistentVolumeClaimModifyVolumeInfeasible) {
+		if pvc.Status.ModifyVolumeStatus != nil && (pvc.Status.ModifyVolumeStatus.Status == v1.PersistentVolumeClaimModifyVolumeInProgress) {
 			pvcKey, err := cache.MetaNamespaceKeyFunc(pvc)
 			if err != nil {
 				return err
 			}
-			ctrl.uncertainPVCs[pvcKey] = *pvc.DeepCopy()
+			ctrl.uncertainPVCs.Store(pvcKey, pvc)
 		}
 	}
 
@@ -160,12 +161,11 @@ func (ctrl *modifyController) updatePVC(oldObj, newObj interface{}) {
 	}
 
 	// Only trigger modify volume if the following conditions are met
-	// 1. Non empty vac name
-	// 2. oldVacName != newVacName
-	// 3. PVC is in Bound state
-	oldVacName := oldPVC.Spec.VolumeAttributesClassName
-	newVacName := newPVC.Spec.VolumeAttributesClassName
-	if newVacName != nil && *newVacName != "" && (oldVacName == nil || *newVacName != *oldVacName) && oldPVC.Status.Phase == v1.ClaimBound {
+	// 1. VAC changed or modify finished (check pending modify request while we are modifying)
+	// 2. PVC is in Bound state
+	oldVacName := ptr.Deref(oldPVC.Spec.VolumeAttributesClassName, "")
+	newVacName := ptr.Deref(newPVC.Spec.VolumeAttributesClassName, "")
+	if (newVacName != oldVacName || newPVC.Status.ModifyVolumeStatus == nil) && newPVC.Status.Phase == v1.ClaimBound {
 		_, err := ctrl.pvLister.Get(oldPVC.Spec.VolumeName)
 		if err != nil {
 			klog.Errorf("Get PV %q of pvc %q in PVInformer cache failed: %v", oldPVC.Spec.VolumeName, klog.KObj(oldPVC), err)
@@ -187,10 +187,7 @@ func (ctrl *modifyController) deletePVC(obj interface{}) {
 }
 
 func (ctrl *modifyController) init(ctx context.Context) bool {
-	informersSyncd := []cache.InformerSynced{ctrl.pvListerSynced, ctrl.pvcListerSynced}
-	informersSyncd = append(informersSyncd, ctrl.vacListerSynced)
-
-	if !cache.WaitForCacheSync(ctx.Done(), informersSyncd...) {
+	if !cache.WaitForCacheSync(ctx.Done(), ctrl.pvListerSynced, ctrl.pvcListerSynced, ctrl.vacListerSynced) {
 		klog.ErrorS(nil, "Cannot sync pod, pv, pvc or vac caches")
 		return false
 	}
@@ -220,7 +217,7 @@ func (ctrl *modifyController) Run(
 	// Starts go-routine that deletes expired slowSet entries.
 	go ctrl.slowSet.Run(stopCh)
 
-	for i := 0; i < workers; i++ {
+	for range workers {
 		go wait.Until(ctrl.sync, 0, stopCh)
 	}
 
@@ -270,15 +267,13 @@ func (ctrl *modifyController) syncPVC(key string) error {
 
 	// Only trigger modify volume if the following conditions are met
 	// 1. PV provisioned by CSI driver AND driver name matches local driver
-	// 2. Non-empty vac name
-	// 3. PVC is in Bound state
+	// 2. PVC is in Bound state
 	if pv.Spec.CSI == nil || pv.Spec.CSI.Driver != ctrl.name {
 		klog.V(7).InfoS("Skipping PV provisioned by different driver", "PV", klog.KObj(pv))
 		return nil
 	}
 
-	vacName := pvc.Spec.VolumeAttributesClassName
-	if vacName != nil && *vacName != "" && pvc.Status.Phase == v1.ClaimBound {
+	if pvc.Status.Phase == v1.ClaimBound {
 		_, _, err, _ := ctrl.modify(pvc, pv)
 		if err != nil {
 			return err