Merge pull request kubernetes#131311 from gnufied/fix-csi-json-file-removal

Fix error handling and csi json file removal interaction

Author: k8s-ci-robot

pkg/volume/csi/csi_attacher.go

@@ -350,25 +350,6 @@ func (c *csiAttacher) MountDevice(spec *volume.Spec, devicePath string, deviceMo
 		data[volDataKey.seLinuxMountContext] = deviceMounterArgs.SELinuxLabel
 	}
 
-	err = saveVolumeData(dataDir, volDataFileName, data)
-	defer func() {
-		// Only if there was an error and volume operation was considered
-		// finished, we should remove the directory.
-		if err != nil && volumetypes.IsOperationFinishedError(err) {
-			// clean up metadata
-			klog.Error(log("attacher.MountDevice failed: %v", err))
-			if err := removeMountDir(c.plugin, deviceMountPath); err != nil {
-				klog.Error(log("attacher.MountDevice failed to remove mount dir after error [%s]: %v", deviceMountPath, err))
-			}
-		}
-	}()
-
-	if err != nil {
-		errMsg := log("failed to save volume info data: %v", err)
-		klog.Error(errMsg)
-		return errors.New(errMsg)
-	}
-
 	if !stageUnstageSet {
 		klog.Info(log("attacher.MountDevice STAGE_UNSTAGE_VOLUME capability not set. Skipping MountDevice..."))
 		// defer does *not* remove the metadata file and it's correct - UnmountDevice needs it there.
@@ -387,13 +368,23 @@ func (c *csiAttacher) MountDevice(spec *volume.Spec, devicePath string, deviceMo
 		return volumetypes.NewTransientOperationFailure(log("attacher.MountDevice failed to determine if the node service has VOLUME_MOUNT_GROUP capability: %v", err))
 	}
 
+	err = saveVolumeData(dataDir, volDataFileName, data)
+	if err != nil {
+		errMsg := log("failed to save volume info data: %v", err)
+		klog.Error(errMsg)
+		if err := removeMountDir(c.plugin, deviceMountPath); err != nil {
+			klog.Error(log("attacher.MountDevice failed to remove mount dir after error [%s]: %v", deviceMountPath, err))
+		}
+		return errors.New(errMsg)
+	}
+
 	if driverSupportsCSIVolumeMountGroup {
 		klog.V(3).Infof("Driver %s supports applying FSGroup (has VOLUME_MOUNT_GROUP node capability). Delegating FSGroup application to the driver through NodeStageVolume.", csiSource.Driver)
 		nodeStageFSGroupArg = deviceMounterArgs.FsGroup
 	}
 
 	fsType := csiSource.FSType
-	err = csi.NodeStageVolume(ctx,
+	csiRPCError := csi.NodeStageVolume(ctx,
 		csiSource.VolumeHandle,
 		publishContext,
 		deviceMountPath,
@@ -404,12 +395,19 @@ func (c *csiAttacher) MountDevice(spec *volume.Spec, devicePath string, deviceMo
 		mountOptions,
 		nodeStageFSGroupArg)
 
-	if err != nil {
-		return err
+	if csiRPCError != nil {
+		if volumetypes.IsOperationFinishedError(csiRPCError) {
+			// clean up metadata
+			klog.Error(log("attacher.MountDevice failed: %v", csiRPCError))
+			if err := removeMountDir(c.plugin, deviceMountPath); err != nil {
+				klog.Error(log("attacher.MountDevice failed to remove mount dir after error [%s]: %v", deviceMountPath, err))
+			}
+		}
+		return csiRPCError
 	}
 
 	klog.V(4).Info(log("attacher.MountDevice successfully requested NodeStageVolume [%s]", deviceMountPath))
-	return err
+	return nil
 }
 
 var _ volume.Detacher = &csiAttacher{}

pkg/volume/csi/csi_mounter.go

@@ -281,23 +281,16 @@ func (c *csiMountMgr) SetUpAt(dir string, mounterArgs volume.MounterArgs) error
 	}
 
 	err = saveVolumeData(parentDir, volDataFileName, volData)
-	defer func() {
-		// Only if there was an error and volume operation was considered
-		// finished, we should remove the directory.
-		if err != nil && volumetypes.IsOperationFinishedError(err) {
-			// attempt to cleanup volume mount dir
-			if removeerr := removeMountDir(c.plugin, dir); removeerr != nil {
-				klog.Error(log("mounter.SetUpAt failed to remove mount dir after error [%s]: %v", dir, removeerr))
-			}
-		}
-	}()
 	if err != nil {
 		errorMsg := log("mounter.SetUpAt failed to save volume info data: %v", err)
 		klog.Error(errorMsg)
-		return volumetypes.NewTransientOperationFailure(errorMsg)
+		if removeerr := removeMountDir(c.plugin, dir); removeerr != nil {
+			klog.Error(log("mounter.SetUpAt failed to remove mount dir after error [%s]: %v", dir, removeerr))
+		}
+		return err
 	}
 
-	err = csi.NodePublishVolume(
+	csiRPCError := csi.NodePublishVolume(
 		ctx,
 		volumeHandle,
 		readOnly,
@@ -312,14 +305,14 @@ func (c *csiMountMgr) SetUpAt(dir string, mounterArgs volume.MounterArgs) error
 		nodePublishFSGroupArg,
 	)
 
-	if err != nil {
+	if csiRPCError != nil {
 		// If operation finished with error then we can remove the mount directory.
-		if volumetypes.IsOperationFinishedError(err) {
+		if volumetypes.IsOperationFinishedError(csiRPCError) {
 			if removeMountDirErr := removeMountDir(c.plugin, dir); removeMountDirErr != nil {
 				klog.Error(log("mounter.SetupAt failed to remove mount dir after a NodePublish() error [%s]: %v", dir, removeMountDirErr))
 			}
 		}
-		return err
+		return csiRPCError
 	}
 
 	if !selinuxLabelMount {
@@ -332,9 +325,13 @@ func (c *csiMountMgr) SetUpAt(dir string, mounterArgs volume.MounterArgs) error
 
 	if !driverSupportsCSIVolumeMountGroup && c.supportsFSGroup(fsType, mounterArgs.FsGroup, fsGroupPolicy) {
 		// Driver doesn't support applying FSGroup. Kubelet must apply it instead.
-
+		var ownershipChanger volume.VolumeOwnershipChanger
+		if mounterArgs.VolumeOwnershipApplicator != nil {
+			ownershipChanger = mounterArgs.VolumeOwnershipApplicator
+		} else {
+			ownershipChanger = volume.NewVolumeOwnership(c, dir, mounterArgs.FsGroup, mounterArgs.FSGroupChangePolicy, util.FSGroupCompleteHook(c.plugin, c.spec))
+		}
 		// fullPluginName helps to distinguish different driver from csi plugin
-		ownershipChanger := volume.NewVolumeOwnership(c, dir, mounterArgs.FsGroup, mounterArgs.FSGroupChangePolicy, util.FSGroupCompleteHook(c.plugin, c.spec))
 		ownershipChanger.AddProgressNotifier(c.pod, mounterArgs.Recorder)
 		err = ownershipChanger.ChangePermissions()
 		if err != nil {

pkg/volume/csi/csi_mounter_test.go

@@ -38,6 +38,7 @@ import (
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	fakeclient "k8s.io/client-go/kubernetes/fake"
 	clitesting "k8s.io/client-go/testing"
+	"k8s.io/client-go/tools/record"
 	featuregatetesting "k8s.io/component-base/featuregate/testing"
 	pkgauthenticationv1 "k8s.io/kubernetes/pkg/apis/authentication/v1"
 	pkgcorev1 "k8s.io/kubernetes/pkg/apis/core/v1"
@@ -360,6 +361,132 @@ func TestMounterSetUp(t *testing.T) {
 	}
 }
 
+type mockVolumeOwnershipChanger struct {
+	triggerError bool
+}
+
+func (m *mockVolumeOwnershipChanger) ChangePermissions() error {
+	if m.triggerError {
+		return fmt.Errorf("mock error")
+	}
+	return nil
+}
+
+func (m *mockVolumeOwnershipChanger) AddProgressNotifier(pod *corev1.Pod, recorder record.EventRecorder) volume.VolumeOwnershipChanger {
+	return m
+}
+
+func TestMounterSetupJsonFileHandling(t *testing.T) {
+	testCases := []struct {
+		name                string
+		volumeID            string
+		setupShouldFail     bool
+		errorType           error
+		failOwnershipChange bool
+		shouldRemoveFile    bool
+	}{
+		{
+			name:             "transient error should not remove json file",
+			volumeID:         fakecsi.NodePublishTimeOut_VolumeID,
+			setupShouldFail:  true,
+			shouldRemoveFile: false,
+		},
+		{
+			name:             "final error should remove json file",
+			volumeID:         fakecsi.NodePublishFinalError_VolumeID,
+			setupShouldFail:  true,
+			shouldRemoveFile: true,
+		},
+		{
+			name:                "error in fsgroup permission change, should not remove json file",
+			volumeID:            testVol,
+			failOwnershipChange: true,
+			setupShouldFail:     true,
+			shouldRemoveFile:    false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			modes := []storage.VolumeLifecycleMode{
+				storage.VolumeLifecyclePersistent,
+			}
+			csiDriver := getTestCSIDriver("file-driver", nil, nil, modes)
+			fileFsPolicy := storage.FileFSGroupPolicy
+			csiDriver.Spec.FSGroupPolicy = &fileFsPolicy
+
+			fakeClient := fakeclient.NewSimpleClientset(csiDriver)
+			plug, tmpDir := newTestPlugin(t, fakeClient)
+			defer os.RemoveAll(tmpDir)
+
+			registerFakePlugin("test-driver", "endpoint", []string{"1.0.0"}, t)
+			pv := makeTestPV("test-vol", 10, "file-driver", tc.volumeID)
+			pv.Spec.MountOptions = []string{"foo=bar", "baz=qux"}
+
+			mounter, err := plug.NewMounter(
+				volume.NewSpecFromPersistentVolume(pv, pv.Spec.PersistentVolumeSource.CSI.ReadOnly),
+				&corev1.Pod{
+					ObjectMeta: meta.ObjectMeta{UID: testPodUID, Namespace: testns, Name: testPod},
+					Spec: corev1.PodSpec{
+						ServiceAccountName: testAccount,
+					},
+				},
+			)
+			if err != nil {
+				t.Fatalf("failed to make a new Mounter: %v", err)
+			}
+
+			if mounter == nil {
+				t.Fatal("failed to create CSI mounter")
+			}
+
+			csiMounter := mounter.(*csiMountMgr)
+			csiMounter.csiClient = setupClient(t, true)
+
+			attachID := getAttachmentName(csiMounter.volumeID, string(csiMounter.driverName), string(plug.host.GetNodeName()))
+			attachment := makeTestAttachment(attachID, "test-node", csiMounter.spec.Name())
+			_, err = csiMounter.k8s.StorageV1().VolumeAttachments().Create(context.TODO(), attachment, meta.CreateOptions{})
+			if err != nil {
+				t.Fatalf("failed to setup VolumeAttachment: %v", err)
+			}
+
+			var mounterArgs volume.MounterArgs
+			fsGroup := int64(2000)
+			mounterArgs.FsGroup = &fsGroup
+			if tc.failOwnershipChange {
+				mounterArgs.VolumeOwnershipApplicator = &mockVolumeOwnershipChanger{triggerError: true}
+			}
+
+			dataDir := filepath.Dir(mounter.GetPath())
+
+			// Mounter.SetUp()
+			err = csiMounter.SetUp(mounterArgs)
+			if tc.setupShouldFail {
+				if err == nil {
+					t.Error("test should fail, but no error occurred")
+				}
+			} else if err != nil {
+				t.Fatal("unexpected error:", err)
+			}
+
+			// Check if the json file exists or not
+			dataFile := filepath.Join(dataDir, volDataFileName)
+			_, err = os.Stat(dataFile)
+			if tc.shouldRemoveFile {
+				if !os.IsNotExist(err) {
+					t.Errorf("Expected json file to be removed, but it still exists: %v", err)
+				}
+			} else {
+				if os.IsNotExist(err) {
+					t.Errorf("Expected json file to exist, but it was removed")
+				} else if err != nil {
+					t.Errorf("Unexpected error while checking json file: %v", err)
+				}
+			}
+		})
+	}
+}
+
 func TestMounterSetUpSimple(t *testing.T) {
 	fakeClient := fakeclient.NewSimpleClientset()
 	plug, tmpDir := newTestPlugin(t, fakeClient)

pkg/volume/csi/fake/fake_client.go

@@ -31,7 +31,8 @@ import (
 
 const (
 	// NodePublishTimeOut_VolumeID is volume id that will result in NodePublish operation to timeout
-	NodePublishTimeOut_VolumeID = "node-publish-timeout"
+	NodePublishTimeOut_VolumeID    = "node-publish-timeout"
+	NodePublishFinalError_VolumeID = "node-publish-final-error"
 
 	// NodeStageTimeOut_VolumeID is a volume id that will result in NodeStage operation to timeout
 	NodeStageTimeOut_VolumeID = "node-stage-timeout"
@@ -206,6 +207,10 @@ func (f *NodeClient) NodePublishVolume(ctx context.Context, req *csipb.NodePubli
 		return nil, timeoutErr
 	}
 
+	if req.GetVolumeId() == NodePublishFinalError_VolumeID {
+		return nil, status.Errorf(codes.Internal, "final error")
+	}
+
 	// "Creation of target_path is the responsibility of the SP."
 	// Our plugin depends on it.
 	if req.VolumeCapability.GetBlock() != nil {

pkg/volume/volume.go

@@ -134,6 +134,15 @@ type MounterArgs struct {
 	DesiredSize         *resource.Quantity
 	SELinuxLabel        string
 	Recorder            record.EventRecorder
+
+	// Optional interface that will be used to change the ownership of the volume, if specified.
+	// mainly used by unit tests
+	VolumeOwnershipApplicator VolumeOwnershipChanger
+}
+
+type VolumeOwnershipChanger interface {
+	AddProgressNotifier(pod *v1.Pod, recorder record.EventRecorder) VolumeOwnershipChanger
+	ChangePermissions() error
 }
 
 type VolumeOwnership struct {

pkg/volume/volume_linux.go

@@ -51,7 +51,7 @@ var (
 )
 
 // NewVolumeOwnership returns an interface that can be used to recursively change volume permissions and ownership
-func NewVolumeOwnership(mounter Mounter, dir string, fsGroup *int64, fsGroupChangePolicy *v1.PodFSGroupChangePolicy, completeFunc func(types.CompleteFuncParam)) *VolumeOwnership {
+func NewVolumeOwnership(mounter Mounter, dir string, fsGroup *int64, fsGroupChangePolicy *v1.PodFSGroupChangePolicy, completeFunc func(types.CompleteFuncParam)) VolumeOwnershipChanger {
 	vo := &VolumeOwnership{
 		mounter:             mounter,
 		dir:                 dir,
@@ -63,7 +63,7 @@ func NewVolumeOwnership(mounter Mounter, dir string, fsGroup *int64, fsGroupChan
 	return vo
 }
 
-func (vo *VolumeOwnership) AddProgressNotifier(pod *v1.Pod, recorder record.EventRecorder) *VolumeOwnership {
+func (vo *VolumeOwnership) AddProgressNotifier(pod *v1.Pod, recorder record.EventRecorder) VolumeOwnershipChanger {
 	vo.pod = pod
 	vo.recorder = recorder
 	return vo

pkg/volume/volume_unsupported.go

@@ -26,11 +26,11 @@ import (
 )
 
 // NewVolumeOwnership returns an interface that can be used to recursively change volume permissions and ownership
-func NewVolumeOwnership(mounter Mounter, dir string, fsGroup *int64, fsGroupChangePolicy *v1.PodFSGroupChangePolicy, completeFunc func(types.CompleteFuncParam)) *VolumeOwnership {
+func NewVolumeOwnership(mounter Mounter, dir string, fsGroup *int64, fsGroupChangePolicy *v1.PodFSGroupChangePolicy, completeFunc func(types.CompleteFuncParam)) VolumeOwnershipChanger {
 	return nil
 }
 
-func (vo *VolumeOwnership) AddProgressNotifier(pod *v1.Pod, recorder record.EventRecorder) *VolumeOwnership {
+func (vo *VolumeOwnership) AddProgressNotifier(pod *v1.Pod, recorder record.EventRecorder) VolumeOwnershipChanger {
 	return vo
 }