Add unit tests for pod information in NodePublish

Some refactoring of CSI tests was necessary.

pkg/volume/csi/BUILD

@@ -47,6 +47,7 @@ go_test(
     ],
     embed = [":go_default_library"],
     deps = [
+        "//pkg/features:go_default_library",
         "//pkg/volume:go_default_library",
         "//pkg/volume/csi/fake:go_default_library",
         "//pkg/volume/testing:go_default_library",
@@ -60,6 +61,7 @@ go_test(
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/watch:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/util/feature/testing:go_default_library",
         "//staging/src/k8s.io/client-go/kubernetes:go_default_library",
         "//staging/src/k8s.io/client-go/kubernetes/fake:go_default_library",
         "//staging/src/k8s.io/client-go/testing:go_default_library",

pkg/volume/csi/csi_attacher_test.go

@@ -747,12 +747,12 @@ func TestAttacherMountDevice(t *testing.T) {
 			t.Errorf("got wrong number of staged volumes, expecting %v got: %v", numStaged, len(staged))
 		}
 		if tc.stageUnstageSet {
-			gotPath, ok := staged[tc.volName]
+			vol, ok := staged[tc.volName]
 			if !ok {
 				t.Errorf("could not find staged volume: %s", tc.volName)
 			}
-			if gotPath != tc.deviceMountPath {
+			if vol.Path != tc.deviceMountPath {
-				t.Errorf("expected mount path: %s. got: %s", tc.deviceMountPath, gotPath)
+				t.Errorf("expected mount path: %s. got: %s", tc.deviceMountPath, vol.Path)
 			}
 		}
 	}
@@ -836,7 +836,7 @@ func TestAttacherUnmountDevice(t *testing.T) {
 
 		// Add the volume to NodeStagedVolumes
 		cdc := csiAttacher.csiClient.(*fakeCsiDriverClient)
-		cdc.nodeClient.AddNodeStagedVolume(tc.volID, tc.deviceMountPath)
+		cdc.nodeClient.AddNodeStagedVolume(tc.volID, tc.deviceMountPath, nil)
 
 		// Make JSON for this object
 		if tc.deviceMountPath != "" {

pkg/volume/csi/csi_block_test.go

@@ -129,9 +129,13 @@ func TestBlockMapperSetupDevice(t *testing.T) {
 	}
 
 	vols := csiMapper.csiClient.(*fakeCsiDriverClient).nodeClient.GetNodeStagedVolumes()
-	if vols[csiMapper.volumeID] != devicePath {
+	vol, ok := vols[csiMapper.volumeID]
+	if !ok {
 		t.Error("csi server may not have received NodePublishVolume call")
 	}
+	if vol.Path != devicePath {
+		t.Errorf("csi server expected device path %s, got %s", devicePath, vol.Path)
+	}
 }
 
 func TestBlockMapperMapDevice(t *testing.T) {
@@ -198,9 +202,13 @@ func TestBlockMapperMapDevice(t *testing.T) {
 	}
 
 	pubs := csiMapper.csiClient.(*fakeCsiDriverClient).nodeClient.GetNodePublishedVolumes()
-	if pubs[csiMapper.volumeID] != podVolumeBlockFilePath {
+	vol, ok := pubs[csiMapper.volumeID]
+	if !ok {
 		t.Error("csi server may not have received NodePublishVolume call")
 	}
+	if vol.Path != podVolumeBlockFilePath {
+		t.Errorf("csi server expected path %s, got %s", podVolumeBlockFilePath, vol.Path)
+	}
 }
 
 func TestBlockMapperTearDownDevice(t *testing.T) {

pkg/volume/csi/csi_mounter.go

@@ -51,6 +51,7 @@ var (
 		"nodeName",
 		"attachmentID",
 	}
+	currentPodInfoMountVersion = "v1"
 )
 
 type csiMountMgr struct {
@@ -253,7 +254,8 @@ func (c *csiMountMgr) podAttributes() (map[string]string, error) {
 		return nil, err
 	}
 
-	if csiDriver.Spec.PodInfoRequiredOnMount == nil || *csiDriver.Spec.PodInfoRequiredOnMount == false {
+	// if PodInfoOnMountVersion is not set or not v1 we do not set pod attributes
+	if csiDriver.Spec.PodInfoOnMountVersion == nil || *csiDriver.Spec.PodInfoOnMountVersion != currentPodInfoMountVersion {
 		glog.V(4).Infof(log("CSIDriver %q does not require pod information", c.driverName))
 		return nil, nil
 	}

pkg/volume/csi/csi_mounter_test.go

@@ -25,14 +25,20 @@ import (
 	"path"
 	"testing"
 
+	"reflect"
+
+	"github.com/golang/glog"
 	api "k8s.io/api/core/v1"
 	storage "k8s.io/api/storage/v1beta1"
 	meta "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	utilfeaturetesting "k8s.io/apiserver/pkg/util/feature/testing"
 	fakeclient "k8s.io/client-go/kubernetes/fake"
 	csiapi "k8s.io/csi-api/pkg/apis/csi/v1alpha1"
+	fakecsi "k8s.io/csi-api/pkg/client/clientset/versioned/fake"
+	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/volume"
-	volumetest "k8s.io/kubernetes/pkg/volume/testing"
 	"k8s.io/kubernetes/pkg/volume/util"
 )
 
@@ -88,86 +94,171 @@ func TestMounterGetPath(t *testing.T) {
 	}
 }
 
+func MounterSetUpTests(t *testing.T, podInfoEnabled bool) {
+	defer utilfeaturetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.CSIPodInfo, podInfoEnabled)()
+	tests := []struct {
+		name       string
+		driver     string
+		attributes map[string]string
+
+		expectedAttributes map[string]string
+	}{
+		{
+			name:               "no pod info",
+			driver:             "no-info",
+			attributes:         nil,
+			expectedAttributes: nil,
+		},
+		{
+			name:               "no CSIDriver -> no pod info",
+			driver:             "unknown-driver",
+			attributes:         nil,
+			expectedAttributes: nil,
+		},
+		{
+			name:               "CSIDriver with PodInfoRequiredOnMount=nil -> no pod info",
+			driver:             "nil",
+			attributes:         nil,
+			expectedAttributes: nil,
+		},
+		{
+			name:               "no pod info -> keep existing attributes",
+			driver:             "no-info",
+			attributes:         map[string]string{"foo": "bar"},
+			expectedAttributes: map[string]string{"foo": "bar"},
+		},
+		{
+			name:               "add pod info",
+			driver:             "info",
+			attributes:         nil,
+			expectedAttributes: map[string]string{"csi.storage.k8s.io/pod.uid": "test-pod", "csi.storage.k8s.io/serviceAccount.name": "test-service-account", "csi.storage.k8s.io/pod.name": "test-pod", "csi.storage.k8s.io/pod.namespace": "test-ns"},
+		},
+		{
+			name:               "add pod info -> keep existing attributes",
+			driver:             "info",
+			attributes:         map[string]string{"foo": "bar"},
+			expectedAttributes: map[string]string{"foo": "bar", "csi.storage.k8s.io/pod.uid": "test-pod", "csi.storage.k8s.io/serviceAccount.name": "test-service-account", "csi.storage.k8s.io/pod.name": "test-pod", "csi.storage.k8s.io/pod.namespace": "test-ns"},
+		},
+	}
+
+	emptyPodMountInfoVersion := ""
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			glog.Infof("Starting test %s", test.name)
+			fakeClient := fakeclient.NewSimpleClientset()
+			fakeCSIClient := fakecsi.NewSimpleClientset(
+				getCSIDriver("no-info", &emptyPodMountInfoVersion, nil),
+				getCSIDriver("info", &currentPodInfoMountVersion, nil),
+				getCSIDriver("nil", nil, nil),
+			)
+			plug, tmpDir := newTestPlugin(t, fakeClient, fakeCSIClient)
+			defer os.RemoveAll(tmpDir)
+
+			for {
+				// Wait until the informer in CSI volume plugin has all CSIDrivers.
+				if plug.csiDriverInformer.Informer().HasSynced() {
+					break
+				}
+			}
+			pv := makeTestPV("test-pv", 10, test.driver, testVol)
+			pv.Spec.CSI.VolumeAttributes = test.attributes
+			pvName := pv.GetName()
+
+			mounter, err := plug.NewMounter(
+				volume.NewSpecFromPersistentVolume(pv, pv.Spec.PersistentVolumeSource.CSI.ReadOnly),
+				&api.Pod{
+					ObjectMeta: meta.ObjectMeta{UID: testPodUID, Namespace: testns, Name: testPod},
+					Spec: api.PodSpec{
+						ServiceAccountName: testAccount,
+					},
+				},
+				volume.VolumeOptions{},
+			)
+			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, csiMounter.driverName, string(plug.host.GetNodeName()))
+
+			attachment := &storage.VolumeAttachment{
+				ObjectMeta: meta.ObjectMeta{
+					Name: attachID,
+				},
+				Spec: storage.VolumeAttachmentSpec{
+					NodeName: "test-node",
+					Attacher: csiPluginName,
+					Source: storage.VolumeAttachmentSource{
+						PersistentVolumeName: &pvName,
+					},
+				},
+				Status: storage.VolumeAttachmentStatus{
+					Attached:    false,
+					AttachError: nil,
+					DetachError: nil,
+				},
+			}
+			_, err = csiMounter.k8s.StorageV1beta1().VolumeAttachments().Create(attachment)
+			if err != nil {
+				t.Fatalf("failed to setup VolumeAttachment: %v", err)
+			}
+
+			// Mounter.SetUp()
+			fsGroup := int64(2000)
+			if err := csiMounter.SetUp(&fsGroup); err != nil {
+				t.Fatalf("mounter.Setup failed: %v", err)
+			}
+
+			//Test the default value of file system type is not overridden
+			if len(csiMounter.spec.PersistentVolume.Spec.CSI.FSType) != 0 {
+				t.Errorf("default value of file system type was overridden by type %s", csiMounter.spec.PersistentVolume.Spec.CSI.FSType)
+			}
+
+			path := csiMounter.GetPath()
+			if _, err := os.Stat(path); err != nil {
+				if os.IsNotExist(err) {
+					t.Errorf("SetUp() failed, volume path not created: %s", path)
+				} else {
+					t.Errorf("SetUp() failed: %v", err)
+				}
+			}
+
+			// ensure call went all the way
+			pubs := csiMounter.csiClient.(*fakeCsiDriverClient).nodeClient.GetNodePublishedVolumes()
+			vol, ok := pubs[csiMounter.volumeID]
+			if !ok {
+				t.Error("csi server may not have received NodePublishVolume call")
+			}
+			if vol.Path != csiMounter.GetPath() {
+				t.Errorf("csi server expected path %s, got %s", csiMounter.GetPath(), vol.Path)
+			}
+			if podInfoEnabled {
+				if !reflect.DeepEqual(vol.Attributes, test.expectedAttributes) {
+					t.Errorf("csi server expected attributes %+v, got %+v", test.expectedAttributes, vol.Attributes)
+				}
+			} else {
+				// CSIPodInfo feature is disabled, we expect no modifications to attributes.
+				if !reflect.DeepEqual(vol.Attributes, test.attributes) {
+					t.Errorf("csi server expected attributes %+v, got %+v", test.attributes, vol.Attributes)
+				}
+			}
+		})
+	}
+}
+
 func TestMounterSetUp(t *testing.T) {
-	plug, tmpDir := newTestPlugin(t, nil, nil)
+	t.Run("WithCSIPodInfo", func(t *testing.T) {
-	defer os.RemoveAll(tmpDir)
+		MounterSetUpTests(t, true)
-	fakeClient := fakeclient.NewSimpleClientset()
+	})
-	host := volumetest.NewFakeVolumeHostWithCSINodeName(
+	t.Run("WithoutCSIPodInfo", func(t *testing.T) {
-		tmpDir,
+		MounterSetUpTests(t, false)
-		fakeClient,
+	})
-		nil,
-		nil,
-		"fakeNode",
-	)
-	plug.host = host
-	pv := makeTestPV("test-pv", 10, testDriver, testVol)
-	pvName := pv.GetName()
-
-	mounter, err := plug.NewMounter(
-		volume.NewSpecFromPersistentVolume(pv, pv.Spec.PersistentVolumeSource.CSI.ReadOnly),
-		&api.Pod{ObjectMeta: meta.ObjectMeta{UID: testPodUID, Namespace: testns}},
-		volume.VolumeOptions{},
-	)
-	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, csiMounter.driverName, string(plug.host.GetNodeName()))
-
-	attachment := &storage.VolumeAttachment{
-		ObjectMeta: meta.ObjectMeta{
-			Name: attachID,
-		},
-		Spec: storage.VolumeAttachmentSpec{
-			NodeName: "test-node",
-			Attacher: csiPluginName,
-			Source: storage.VolumeAttachmentSource{
-				PersistentVolumeName: &pvName,
-			},
-		},
-		Status: storage.VolumeAttachmentStatus{
-			Attached:    false,
-			AttachError: nil,
-			DetachError: nil,
-		},
-	}
-	_, err = csiMounter.k8s.StorageV1beta1().VolumeAttachments().Create(attachment)
-	if err != nil {
-		t.Fatalf("failed to setup VolumeAttachment: %v", err)
-	}
-
-	// Mounter.SetUp()
-	fsGroup := int64(2000)
-	if err := csiMounter.SetUp(&fsGroup); err != nil {
-		t.Fatalf("mounter.Setup failed: %v", err)
-	}
-
-	//Test the default value of file system type is not overridden
-	if len(csiMounter.spec.PersistentVolume.Spec.CSI.FSType) != 0 {
-		t.Errorf("default value of file system type was overridden by type %s", csiMounter.spec.PersistentVolume.Spec.CSI.FSType)
-	}
-
-	path := csiMounter.GetPath()
-	if _, err := os.Stat(path); err != nil {
-		if os.IsNotExist(err) {
-			t.Errorf("SetUp() failed, volume path not created: %s", path)
-		} else {
-			t.Errorf("SetUp() failed: %v", err)
-		}
-	}
-
-	// ensure call went all the way
-	pubs := csiMounter.csiClient.(*fakeCsiDriverClient).nodeClient.GetNodePublishedVolumes()
-	if pubs[csiMounter.volumeID] != csiMounter.GetPath() {
-		t.Error("csi server may not have received NodePublishVolume call")
-	}
 }
 
 func TestUnmounterTeardown(t *testing.T) {
@@ -267,14 +358,13 @@ func TestSaveVolumeData(t *testing.T) {
 	}
 }
 
-func getCSIDriver(name string, requiresPodInfo *bool, attachable *bool) *csiapi.CSIDriver {
+func getCSIDriver(name string, podInfoMountVersion *string, attachable *bool) *csiapi.CSIDriver {
-	podInfoMountVersion := "v1"
 	return &csiapi.CSIDriver{
 		ObjectMeta: meta.ObjectMeta{
 			Name: name,
 		},
 		Spec: csiapi.CSIDriverSpec{
-			PodInfoOnMountVersion: &podInfoMountVersion,
+			PodInfoOnMountVersion: podInfoMountVersion,
 			AttachRequired:        attachable,
 		},
 	}

pkg/volume/csi/fake/fake_client.go

@@ -56,10 +56,15 @@ func (f *IdentityClient) Probe(ctx context.Context, in *csipb.ProbeRequest, opts
 	return nil, nil
 }
 
+type CSIVolume struct {
+	Attributes map[string]string
+	Path       string
+}
+
 // NodeClient returns CSI node client
 type NodeClient struct {
-	nodePublishedVolumes map[string]string
+	nodePublishedVolumes map[string]CSIVolume
-	nodeStagedVolumes    map[string]string
+	nodeStagedVolumes    map[string]CSIVolume
 	stageUnstageSet      bool
 	nodeGetInfoResp      *csipb.NodeGetInfoResponse
 	nextErr              error
@@ -68,8 +73,8 @@ type NodeClient struct {
 // NewNodeClient returns fake node client
 func NewNodeClient(stageUnstageSet bool) *NodeClient {
 	return &NodeClient{
-		nodePublishedVolumes: make(map[string]string),
+		nodePublishedVolumes: make(map[string]CSIVolume),
-		nodeStagedVolumes:    make(map[string]string),
+		nodeStagedVolumes:    make(map[string]CSIVolume),
 		stageUnstageSet:      stageUnstageSet,
 	}
 }
@@ -84,17 +89,20 @@ func (f *NodeClient) SetNodeGetInfoResp(resp *csipb.NodeGetInfoResponse) {
 }
 
 // GetNodePublishedVolumes returns node published volumes
-func (f *NodeClient) GetNodePublishedVolumes() map[string]string {
+func (f *NodeClient) GetNodePublishedVolumes() map[string]CSIVolume {
 	return f.nodePublishedVolumes
 }
 
 // GetNodeStagedVolumes returns node staged volumes
-func (f *NodeClient) GetNodeStagedVolumes() map[string]string {
+func (f *NodeClient) GetNodeStagedVolumes() map[string]CSIVolume {
 	return f.nodeStagedVolumes
 }
 
-func (f *NodeClient) AddNodeStagedVolume(volID, deviceMountPath string) {
+func (f *NodeClient) AddNodeStagedVolume(volID, deviceMountPath string, attributes map[string]string) {
-	f.nodeStagedVolumes[volID] = deviceMountPath
+	f.nodeStagedVolumes[volID] = CSIVolume{
+		Path:       deviceMountPath,
+		Attributes: attributes,
+	}
 }
 
 // NodePublishVolume implements CSI NodePublishVolume
@@ -115,7 +123,10 @@ func (f *NodeClient) NodePublishVolume(ctx context.Context, req *csipb.NodePubli
 	if !strings.Contains(fsTypes, fsType) {
 		return nil, errors.New("invalid fstype")
 	}
-	f.nodePublishedVolumes[req.GetVolumeId()] = req.GetTargetPath()
+	f.nodePublishedVolumes[req.GetVolumeId()] = CSIVolume{
+		Path:       req.GetTargetPath(),
+		Attributes: req.GetVolumeAttributes(),
+	}
 	return &csipb.NodePublishVolumeResponse{}, nil
 }
 
@@ -158,7 +169,10 @@ func (f *NodeClient) NodeStageVolume(ctx context.Context, req *csipb.NodeStageVo
 		return nil, errors.New("invalid fstype")
 	}
 
-	f.nodeStagedVolumes[req.GetVolumeId()] = req.GetStagingTargetPath()
+	f.nodeStagedVolumes[req.GetVolumeId()] = CSIVolume{
+		Path:       req.GetStagingTargetPath(),
+		Attributes: req.GetVolumeAttributes(),
+	}
 	return &csipb.NodeStageVolumeResponse{}, nil
 }
 

test/integration/volume/attach_detach_test.go

@@ -415,7 +415,7 @@ func createAdClients(ns *v1.Namespace, t *testing.T, server *httptest.Server, sy
 	ctrl, err := attachdetach.NewAttachDetachController(
 		testClient,
 		nil, /* csiClient */
-		nil, /* crdClient */
+		fakeApiExtensionsClient, /* crdClient */
 		informers.Core().V1().Pods(),
 		informers.Core().V1().Nodes(),
 		informers.Core().V1().PersistentVolumeClaims(),