Add tests for verifying in-progress state

pkg/volume/csi/csi_attacher.go

@@ -309,6 +309,11 @@ func (c *csiAttacher) MountDeviceWithStatusTracking(spec *volume.Spec, devicePat
 	nodeName := string(c.plugin.host.GetNodeName())
 	publishContext, err := c.plugin.getPublishContext(c.k8s, csiSource.VolumeHandle, csiSource.Driver, nodeName)
 
+	if err != nil {
+		opExitStatus = volumetypes.OperationStateNoChange
+		return opExitStatus, err
+	}
+
 	nodeStageSecrets := map[string]string{}
 	if csiSource.NodeStageSecretRef != nil {
 		nodeStageSecrets, err = getCredentialsFromSecret(c.k8s, csiSource.NodeStageSecretRef)

pkg/volume/csi/csi_attacher_test.go

@@ -44,7 +44,9 @@ import (
 	featuregatetesting "k8s.io/component-base/featuregate/testing"
 	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/volume"
+	fakecsi "k8s.io/kubernetes/pkg/volume/csi/fake"
 	volumetest "k8s.io/kubernetes/pkg/volume/testing"
+	volumetypes "k8s.io/kubernetes/pkg/volume/util/types"
 )
 
 var (
@@ -1055,72 +1057,110 @@ func TestAttacherGetDeviceMountPath(t *testing.T) {
 func TestAttacherMountDevice(t *testing.T) {
 	pvName := "test-pv"
 	testCases := []struct {
-		testName        string
+		testName         string
-		volName         string
+		volName          string
-		devicePath      string
+		devicePath       string
-		deviceMountPath string
+		deviceMountPath  string
-		stageUnstageSet bool
+		stageUnstageSet  bool
-		shouldFail      bool
+		shouldFail       bool
-		spec            *volume.Spec
+		createAttachment bool
+		exitStatus       volumetypes.OperationStatus
+		spec             *volume.Spec
 	}{
 		{
-			testName:        "normal PV",
+			testName:         "normal PV",
-			volName:         "test-vol1",
+			volName:          "test-vol1",
-			devicePath:      "path1",
+			devicePath:       "path1",
-			deviceMountPath: "path2",
+			deviceMountPath:  "path2",
-			stageUnstageSet: true,
+			stageUnstageSet:  true,
-			spec:            volume.NewSpecFromPersistentVolume(makeTestPV(pvName, 10, testDriver, "test-vol1"), false),
+			createAttachment: true,
+			spec:             volume.NewSpecFromPersistentVolume(makeTestPV(pvName, 10, testDriver, "test-vol1"), false),
+			exitStatus:       volumetypes.OperationFinished,
 		},
 		{
-			testName:        "normal PV with mount options",
+			testName:         "normal PV with mount options",
-			volName:         "test-vol1",
+			volName:          "test-vol1",
-			devicePath:      "path1",
+			devicePath:       "path1",
-			deviceMountPath: "path2",
+			deviceMountPath:  "path2",
-			stageUnstageSet: true,
+			stageUnstageSet:  true,
-			spec:            volume.NewSpecFromPersistentVolume(makeTestPVWithMountOptions(pvName, 10, testDriver, "test-vol1", []string{"test-op"}), false),
+			createAttachment: true,
+			exitStatus:       volumetypes.OperationFinished,
+			spec:             volume.NewSpecFromPersistentVolume(makeTestPVWithMountOptions(pvName, 10, testDriver, "test-vol1", []string{"test-op"}), false),
 		},
 		{
-			testName:        "no vol name",
+			testName:         "normal PV but with missing attachment should result in no-change",
-			volName:         "",
+			volName:          "test-vol1",
-			devicePath:      "path1",
+			devicePath:       "path1",
-			deviceMountPath: "path2",
+			deviceMountPath:  "path2",
-			stageUnstageSet: true,
+			stageUnstageSet:  true,
-			shouldFail:      true,
+			createAttachment: false,
-			spec:            volume.NewSpecFromPersistentVolume(makeTestPV(pvName, 10, testDriver, ""), false),
+			shouldFail:       true,
+			exitStatus:       volumetypes.OperationStateNoChange,
+			spec:             volume.NewSpecFromPersistentVolume(makeTestPVWithMountOptions(pvName, 10, testDriver, "test-vol1", []string{"test-op"}), false),
 		},
 		{
-			testName:        "no device path",
+			testName:         "no vol name",
-			volName:         "test-vol1",
+			volName:          "",
-			devicePath:      "",
+			devicePath:       "path1",
-			deviceMountPath: "path2",
+			deviceMountPath:  "path2",
-			stageUnstageSet: true,
+			stageUnstageSet:  true,
-			shouldFail:      false,
+			shouldFail:       true,
-			spec:            volume.NewSpecFromPersistentVolume(makeTestPV(pvName, 10, testDriver, "test-vol1"), false),
+			createAttachment: true,
+			exitStatus:       volumetypes.OperationFinished,
+			spec:             volume.NewSpecFromPersistentVolume(makeTestPV(pvName, 10, testDriver, ""), false),
 		},
 		{
-			testName:        "no device mount path",
+			testName:         "no device path",
-			volName:         "test-vol1",
+			volName:          "test-vol1",
-			devicePath:      "path1",
+			devicePath:       "",
-			deviceMountPath: "",
+			deviceMountPath:  "path2",
-			stageUnstageSet: true,
+			stageUnstageSet:  true,
-			shouldFail:      true,
+			shouldFail:       false,
-			spec:            volume.NewSpecFromPersistentVolume(makeTestPV(pvName, 10, testDriver, "test-vol1"), false),
+			createAttachment: true,
+			exitStatus:       volumetypes.OperationFinished,
+			spec:             volume.NewSpecFromPersistentVolume(makeTestPV(pvName, 10, testDriver, "test-vol1"), false),
 		},
 		{
-			testName:        "stage_unstage cap not set",
+			testName:         "no device mount path",
-			volName:         "test-vol1",
+			volName:          "test-vol1",
-			devicePath:      "path1",
+			devicePath:       "path1",
-			deviceMountPath: "path2",
+			deviceMountPath:  "",
-			stageUnstageSet: false,
+			stageUnstageSet:  true,
-			spec:            volume.NewSpecFromPersistentVolume(makeTestPV(pvName, 10, testDriver, "test-vol1"), false),
+			shouldFail:       true,
+			createAttachment: true,
+			exitStatus:       volumetypes.OperationFinished,
+			spec:             volume.NewSpecFromPersistentVolume(makeTestPV(pvName, 10, testDriver, "test-vol1"), false),
 		},
 		{
-			testName:        "failure with volume source",
+			testName:         "stage_unstage cap not set",
-			volName:         "test-vol1",
+			volName:          "test-vol1",
-			devicePath:      "path1",
+			devicePath:       "path1",
-			deviceMountPath: "path2",
+			deviceMountPath:  "path2",
-			shouldFail:      true,
+			stageUnstageSet:  false,
-			spec:            volume.NewSpecFromVolume(makeTestVol(pvName, testDriver)),
+			createAttachment: true,
+			exitStatus:       volumetypes.OperationFinished,
+			spec:             volume.NewSpecFromPersistentVolume(makeTestPV(pvName, 10, testDriver, "test-vol1"), false),
+		},
+		{
+			testName:         "failure with volume source",
+			volName:          "test-vol1",
+			devicePath:       "path1",
+			deviceMountPath:  "path2",
+			shouldFail:       true,
+			createAttachment: true,
+			exitStatus:       volumetypes.OperationFinished,
+			spec:             volume.NewSpecFromVolume(makeTestVol(pvName, testDriver)),
+		},
+		{
+			testName:         "pv with nodestage timeout should result in in-progress device",
+			volName:          fakecsi.NodeStageTimeOut_VolumeID,
+			devicePath:       "path1",
+			deviceMountPath:  "path2",
+			stageUnstageSet:  true,
+			createAttachment: true,
+			spec:             volume.NewSpecFromPersistentVolume(makeTestPV(pvName, 10, testDriver, fakecsi.NodeStageTimeOut_VolumeID), false),
+			exitStatus:       volumetypes.OperationInProgress,
+			shouldFail:       true,
 		},
 	}
 
@@ -1146,18 +1186,20 @@ func TestAttacherMountDevice(t *testing.T) {
 			nodeName := string(csiAttacher.plugin.host.GetNodeName())
 			attachID := getAttachmentName(tc.volName, testDriver, nodeName)
 
-			// Set up volume attachment
+			if tc.createAttachment {
-			attachment := makeTestAttachment(attachID, nodeName, pvName)
+				// Set up volume attachment
-			_, err := csiAttacher.k8s.StorageV1().VolumeAttachments().Create(attachment)
+				attachment := makeTestAttachment(attachID, nodeName, pvName)
-			if err != nil {
+				_, err := csiAttacher.k8s.StorageV1().VolumeAttachments().Create(attachment)
-				t.Fatalf("failed to attach: %v", err)
+				if err != nil {
+					t.Fatalf("failed to attach: %v", err)
+				}
+				go func() {
+					fakeWatcher.Delete(attachment)
+				}()
 			}
-			go func() {
-				fakeWatcher.Delete(attachment)
-			}()
 
 			// Run
-			err = csiAttacher.MountDevice(tc.spec, tc.devicePath, tc.deviceMountPath)
+			exitStatus, err := csiAttacher.MountDeviceWithStatusTracking(tc.spec, tc.devicePath, tc.deviceMountPath)
 
 			// Verify
 			if err != nil {
@@ -1170,6 +1212,10 @@ func TestAttacherMountDevice(t *testing.T) {
 				t.Errorf("test should fail, but no error occurred")
 			}
 
+			if exitStatus != tc.exitStatus {
+				t.Fatalf("expected exitStatus: %v got: %v", tc.exitStatus, exitStatus)
+			}
+
 			// Verify call goes through all the way
 			numStaged := 1
 			if !tc.stageUnstageSet {

pkg/volume/csi/csi_client_test.go

@@ -28,6 +28,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/kubernetes/pkg/volume"
 	"k8s.io/kubernetes/pkg/volume/csi/fake"
+	volumetypes "k8s.io/kubernetes/pkg/volume/util/types"
 )
 
 type fakeCsiDriverClient struct {
@@ -156,6 +157,12 @@ func (c *fakeCsiDriverClient) NodePublishVolume(
 	}
 
 	_, err := c.nodeClient.NodePublishVolume(ctx, req)
+	if err != nil {
+		if isFinalError(err) {
+			return err
+		}
+		return volumetypes.NewOperationTimedOutError(err.Error())
+	}
 	return err
 }
 
@@ -201,6 +208,12 @@ func (c *fakeCsiDriverClient) NodeStageVolume(ctx context.Context,
 	}
 
 	_, err := c.nodeClient.NodeStageVolume(ctx, req)
+	if err != nil {
+		if isFinalError(err) {
+			return err
+		}
+		return volumetypes.NewOperationTimedOutError(err.Error())
+	}
 	return err
 }
 

pkg/volume/csi/csi_mounter.go

@@ -104,8 +104,8 @@ func (c *csiMountMgr) SetUp(mounterArgs volume.MounterArgs) error {
 }
 
 func (c *csiMountMgr) SetUpWithStatusTracking(mounterArgs volume.MounterArgs) (volumetypes.OperationStatus, error) {
-	err := c.SetUp(mounterArgs)
+	opExitStatus, err := c.setupUtil(c.GetPath(), mounterArgs)
-	return volumetypes.OperationFinished, err
+	return opExitStatus, err
 }
 
 func (c *csiMountMgr) SetUpAt(dir string, mounterArgs volume.MounterArgs) error {

pkg/volume/csi/csi_mounter_test.go

@@ -36,7 +36,9 @@ import (
 	featuregatetesting "k8s.io/component-base/featuregate/testing"
 	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/volume"
+	fakecsi "k8s.io/kubernetes/pkg/volume/csi/fake"
 	"k8s.io/kubernetes/pkg/volume/util"
+	volumetypes "k8s.io/kubernetes/pkg/volume/util/types"
 )
 
 var (
@@ -396,6 +398,113 @@ func TestMounterSetUpSimple(t *testing.T) {
 	}
 }
 
+func TestMounterSetupWithStatusTracking(t *testing.T) {
+	fakeClient := fakeclient.NewSimpleClientset()
+	plug, tmpDir := newTestPlugin(t, fakeClient)
+	defer os.RemoveAll(tmpDir)
+
+	testCases := []struct {
+		name             string
+		podUID           types.UID
+		spec             func(string, []string) *volume.Spec
+		shouldFail       bool
+		exitStatus       volumetypes.OperationStatus
+		createAttachment bool
+	}{
+		{
+			name:   "setup with correct persistent volume source should result in finish exit status",
+			podUID: types.UID(fmt.Sprintf("%08X", rand.Uint64())),
+			spec: func(fsType string, options []string) *volume.Spec {
+				pvSrc := makeTestPV("pv1", 20, testDriver, "vol1")
+				pvSrc.Spec.CSI.FSType = fsType
+				pvSrc.Spec.MountOptions = options
+				return volume.NewSpecFromPersistentVolume(pvSrc, false)
+			},
+			exitStatus:       volumetypes.OperationFinished,
+			createAttachment: true,
+		},
+		{
+			name:   "setup with missing attachment should result in nochange",
+			podUID: types.UID(fmt.Sprintf("%08X", rand.Uint64())),
+			spec: func(fsType string, options []string) *volume.Spec {
+				return volume.NewSpecFromPersistentVolume(makeTestPV("pv3", 20, testDriver, "vol4"), false)
+			},
+			exitStatus:       volumetypes.OperationStateNoChange,
+			createAttachment: false,
+			shouldFail:       true,
+		},
+		{
+			name:   "setup with timeout errors on NodePublish",
+			podUID: types.UID(fmt.Sprintf("%08X", rand.Uint64())),
+			spec: func(fsType string, options []string) *volume.Spec {
+				return volume.NewSpecFromPersistentVolume(makeTestPV("pv4", 20, testDriver, fakecsi.NodePublishTimeOut_VolumeID), false)
+			},
+			createAttachment: true,
+			exitStatus:       volumetypes.OperationInProgress,
+			shouldFail:       true,
+		},
+		{
+			name:   "setup with missing secrets should result in nochange exit",
+			podUID: types.UID(fmt.Sprintf("%08X", rand.Uint64())),
+			spec: func(fsType string, options []string) *volume.Spec {
+				pv := makeTestPV("pv5", 20, testDriver, "vol6")
+				pv.Spec.PersistentVolumeSource.CSI.NodePublishSecretRef = &api.SecretReference{
+					Name:      "foo",
+					Namespace: "default",
+				}
+				return volume.NewSpecFromPersistentVolume(pv, false)
+			},
+			exitStatus:       volumetypes.OperationStateNoChange,
+			createAttachment: true,
+			shouldFail:       true,
+		},
+	}
+
+	for _, tc := range testCases {
+		registerFakePlugin(testDriver, "endpoint", []string{"1.0.0"}, t)
+		t.Run(tc.name, func(t *testing.T) {
+			mounter, err := plug.NewMounter(
+				tc.spec("ext4", []string{}),
+				&api.Pod{ObjectMeta: meta.ObjectMeta{UID: tc.podUID, Namespace: testns}},
+				volume.VolumeOptions{},
+			)
+			if mounter == nil {
+				t.Fatal("failed to create CSI mounter")
+			}
+
+			csiMounter := mounter.(*csiMountMgr)
+			csiMounter.csiClient = setupClient(t, true)
+
+			if csiMounter.volumeLifecycleMode != storagev1beta1.VolumeLifecyclePersistent {
+				t.Fatal("unexpected volume mode: ", csiMounter.volumeLifecycleMode)
+			}
+
+			if tc.createAttachment {
+				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(attachment)
+				if err != nil {
+					t.Fatalf("failed to setup VolumeAttachment: %v", err)
+				}
+			}
+
+			opExistStatus, err := csiMounter.SetUpWithStatusTracking(volume.MounterArgs{})
+
+			if opExistStatus != tc.exitStatus {
+				t.Fatalf("expected exitStatus: %v but got %v", tc.exitStatus, opExistStatus)
+			}
+
+			if tc.shouldFail && err == nil {
+				t.Fatalf("expected failure but Setup succeeded")
+			}
+
+			if !tc.shouldFail && err != nil {
+				t.Fatalf("expected successs got mounter.Setup failed with: %v", err)
+			}
+		})
+	}
+}
+
 func TestMounterSetUpWithInline(t *testing.T) {
 	defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.CSIInlineVolume, true)()
 

pkg/volume/csi/fake/fake_client.go

@@ -21,9 +21,17 @@ import (
 	"errors"
 	"strings"
 
-	"google.golang.org/grpc"
-
 	csipb "github.com/container-storage-interface/spec/lib/go/csi"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+const (
+	// NodePublishTimeout_VolumeID is volume id that will result in NodePublish operation to timeout
+	NodePublishTimeOut_VolumeID = "node-publish-timeout"
+	// NodeStageTimeOut_VolumeID is a volume id that will result in NodeStage operation to timeout
+	NodeStageTimeOut_VolumeID = "node-stage-timeout"
 )
 
 // IdentityClient is a CSI identity client used for testing
@@ -158,6 +166,12 @@ func (f *NodeClient) NodePublishVolume(ctx context.Context, req *csipb.NodePubli
 	if !strings.Contains(fsTypes, fsType) {
 		return nil, errors.New("invalid fstype")
 	}
+
+	if req.GetVolumeId() == NodePublishTimeOut_VolumeID {
+		timeoutErr := status.Errorf(codes.DeadlineExceeded, "timeout exceeded")
+		return nil, timeoutErr
+	}
+
 	f.nodePublishedVolumes[req.GetVolumeId()] = CSIVolume{
 		VolumeHandle:    req.GetVolumeId(),
 		Path:            req.GetTargetPath(),
@@ -214,6 +228,11 @@ func (f *NodeClient) NodeStageVolume(ctx context.Context, req *csipb.NodeStageVo
 		return nil, errors.New("invalid fstype")
 	}
 
+	if req.GetVolumeId() == NodeStageTimeOut_VolumeID {
+		timeoutErr := status.Errorf(codes.DeadlineExceeded, "timeout exceeded")
+		return nil, timeoutErr
+	}
+
 	f.nodeStagedVolumes[req.GetVolumeId()] = csiVol
 	return &csipb.NodeStageVolumeResponse{}, nil
 }