Enable CSI ephemeral volumes support. (#97)

awels · web-flow · commit d69e59ebf79e · 2021-12-17T16:42:20.000+01:00
Enabled CSI ephemeral volumes to work with hpp

Signed-off-by: Alexander Wels &lt;awels@redhat.com&gt;
diff --git a/hack/test-driver.yaml b/hack/test-driver.yaml
@@ -19,3 +19,6 @@ DriverInfo:
     capacity: true
   TopologyKeys:
   - topology.hostpath.csi/node
+InlineVolumes:
+- shared: true
+
diff --git a/pkg/hostpath/controllerserver.go b/pkg/hostpath/controllerserver.go
@@ -18,9 +18,7 @@ package hostpath
 
 import (
 	"fmt"
-	"io/ioutil"
 	"path/filepath"
-	"sort"
 
 	"github.com/golang/protobuf/ptypes/wrappers"
 	"golang.org/x/net/context"
@@ -172,7 +170,7 @@ func (hpc *hostPathController) DeleteVolume(ctx context.Context, req *csi.Delete
 	}
 	if volumePath != "" {
 		if err := DeleteVolume(filepath.Dir(volumePath), req.GetVolumeId()); err != nil {
-			return nil, fmt.Errorf("failed to delete volume %v: %w", req.GetVolumeId(), err)
+			return nil, fmt.Errorf("failed to delete volume %s: %v", req.GetVolumeId(), err)
 		}
 		klog.V(4).Infof("volume %v successfully deleted", req.GetVolumeId())
 	}
@@ -245,20 +243,7 @@ func (hpc *hostPathController) GetCapacity(ctx context.Context, req *csi.GetCapa
 }
 
 func (hpc *hostPathController) getVolumeDirectories() ([]string, error) {
-	directories := make([]string, 0)
-	for _, path := range hpc.cfg.StoragePoolDataDir {
-		files, err := ioutil.ReadDir(path)
-		if err != nil {
-			return nil, err
-		}
-		for _, file := range files {
-			if file.IsDir() {
-				directories = append(directories, filepath.Join(path, file.Name()))
-			}
-		}
-	}
-	sort.Strings(directories)
-	return directories, nil
+	return getVolumeDirectories(hpc.cfg.StoragePoolDataDir)
 }
 
 func (hpc *hostPathController) validateListVolumesRequest(req *csi.ListVolumesRequest) error {
diff --git a/pkg/hostpath/hostpath.go b/pkg/hostpath/hostpath.go
@@ -39,6 +39,7 @@ type Config struct {
 	Endpoint              string
 	NodeID                string
 	StoragePoolDataDir			map[string]string
+	DefaultStoragePoolName string
 	Version	       		  string
 	Mounter mount.Interface
 }
@@ -68,13 +69,16 @@ func NewHostPathDriver(cfg *Config, dataDir string) (*hostPath, error) {
 	if cfg.Mounter == nil {
 		cfg.Mounter = mount.New("")
 	}
-	cfg.StoragePoolDataDir = make(map[string]string, 0)
+	cfg.StoragePoolDataDir = make(map[string]string)
 
 	storagePools := make([]StoragePoolInfo, 0)
 	if err := json.Unmarshal([]byte(dataDir), &storagePools); err != nil {
 		return nil, errors.New("unable to parse storage pool info")
 	}
 	for _, storagePool := range storagePools {
+		if len(cfg.DefaultStoragePoolName) == 0 {
+			cfg.DefaultStoragePoolName = storagePool.Name
+		}
 		cfg.StoragePoolDataDir[storagePool.Name] = storagePool.Path
 	}
 
diff --git a/pkg/hostpath/nodeserver.go b/pkg/hostpath/nodeserver.go
@@ -29,7 +29,10 @@ import (
 	"k8s.io/utils/mount"
 )
 
-const TopologyKeyNode = "topology.hostpath.csi/node"
+const (
+	TopologyKeyNode = "topology.hostpath.csi/node"
+	ephemeralContextKey = "csi.storage.k8s.io/ephemeral"
+)
 
 type hostPathNode struct {
 	cfg *Config
@@ -135,7 +138,15 @@ func (hpn *hostPathNode) mountVolume(targetPath string, req *csi.NodePublishVolu
 	storagePoolName := getStoragePoolNameFromMap(req.GetVolumeContext())
 
 	mounter := hpn.cfg.Mounter
-	path := filepath.Join(hpn.cfg.StoragePoolDataDir[storagePoolName], volumeId)
+	path := ""
+	if isEphemeralVolumeRequest(req) {
+		path = hpn.getEphemeralVolumePath(storagePoolName, volumeId)
+		if err := CreateVolume(filepath.Dir(path), volumeId); err != nil {
+			return fmt.Errorf("failed to create ephemeral volume %v: %w", volumeId, err)
+		}
+	} else {
+		path = filepath.Join(hpn.cfg.StoragePoolDataDir[storagePoolName], volumeId)
+	}
 
 	if err := mounter.Mount(path, targetPath, fsType, options); err != nil {
 		var errList strings.Builder
@@ -145,11 +156,33 @@ func (hpn *hostPathNode) mountVolume(targetPath string, req *csi.NodePublishVolu
 			errList.WriteString(err.Error())
 		}
 		errList.WriteString(fmt.Sprintf("%v", fileInfo.Mode()))
+		if isEphemeralVolumeRequest(req) {
+			if rmErr := os.RemoveAll(path); rmErr != nil && !os.IsNotExist(rmErr) {
+				errList.WriteString(fmt.Sprintf(" :%s", rmErr.Error()))
+			}
+		}
+
 		return fmt.Errorf("failed to mount device: %s at %s: %s", path, targetPath, errList.String())
 	}
 	return nil
 }
 
+func isEphemeralVolumeRequest(req *csi.NodePublishVolumeRequest) bool {
+	return req.GetVolumeContext()[ephemeralContextKey] == "true"
+}
+
+func isEphemeralVolumeId(volumeId string) bool {
+	return strings.HasPrefix(volumeId, "csi")
+}
+
+func (hpn *hostPathNode) getEphemeralVolumePath(storagePoolName, volumeId string) string {
+	storagePoolPath := hpn.cfg.StoragePoolDataDir[storagePoolName]
+	if len(storagePoolPath) == 0 {
+		storagePoolPath = hpn.cfg.StoragePoolDataDir[hpn.cfg.DefaultStoragePoolName]
+	}
+	return filepath.Join(storagePoolPath, volumeId)
+}
+
 func (hpn *hostPathNode) validateNodeUnpublishRequest(req *csi.NodeUnpublishVolumeRequest) error {
 	// Check arguments
 	if len(req.GetVolumeId()) == 0 {
@@ -162,6 +195,9 @@ func (hpn *hostPathNode) validateNodeUnpublishRequest(req *csi.NodeUnpublishVolu
 }
 
 func (hpn *hostPathNode) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpublishVolumeRequest) (*csi.NodeUnpublishVolumeResponse, error) {
+	if req != nil {
+		klog.V(3).Infof("Node Unpublish Request: %+v", *req)
+	}
 	if err := hpn.validateNodeUnpublishRequest(req); err != nil {
 		return nil, err
 	}
@@ -188,10 +224,38 @@ func (hpn *hostPathNode) NodeUnpublishVolume(ctx context.Context, req *csi.NodeU
 		return nil, fmt.Errorf("remove target path: %w", err)
 	}
 	klog.V(4).Infof("hostpath: volume %s has been unpublished.", targetPath)
-
+	if isEphemeralVolumeId(req.GetVolumeId()) {
+		if err := hpn.removeEphemeralPath(req.GetVolumeId()); err != nil {
+			return nil, fmt.Errorf("failed to delete ephemeral volume: %s, %v", req.GetVolumeId(), err)
+		}
+	}
 	return &csi.NodeUnpublishVolumeResponse{}, nil
 }
 
+func (hpn *hostPathNode) removeEphemeralPath(volumeId string) error {
+	volumeDirs, err := hpn.getVolumeDirectories()
+	if err != nil {
+		return err
+	}
+	volumePath := ""
+	for _, volumeDir := range volumeDirs {
+		if filepath.Base(volumeDir) == volumeId {
+			volumePath = volumeDir
+		}
+	}
+	if volumePath != "" {
+		if err := DeleteVolume(filepath.Dir(volumePath), volumeId); err != nil {
+			return fmt.Errorf("failed to delete ephemeral volume %s: %v", volumeId, err)
+		}
+		klog.V(4).Infof("ephemeral volume %v successfully deleted", volumeId)
+	}
+	return nil
+}
+
+func (hpn *hostPathNode) getVolumeDirectories() ([]string, error) {
+	return getVolumeDirectories(hpn.cfg.StoragePoolDataDir)
+}
+
 func (hpn *hostPathNode) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRequest) (*csi.NodeStageVolumeResponse, error) {
 	return nil, status.Error(codes.Unimplemented, "NodeStageVolume is not supported")
 }
diff --git a/pkg/hostpath/nodeserver_test.go b/pkg/hostpath/nodeserver_test.go
@@ -387,6 +387,64 @@ func Test_NodePublishVolumeValidMountPoint(t *testing.T) {
 	Expect(mountPoint.Opts).To(ContainElement("bind"))
 }
 
+func Test_NodePublishVolumeEphemeral(t *testing.T) {
+	RegisterTestingT(t)
+	tempDir, err := ioutil.TempDir(os.TempDir(), "")
+	Expect(err).ToNot(HaveOccurred())
+	defer os.RemoveAll(tempDir)
+	nodeServer := createNodeServer(testNode)
+	fakeMounter := mount.NewFakeMounter([]mount.MountPoint{})
+	nodeServer.cfg.Mounter = fakeMounter
+	nodeServer.cfg.DefaultStoragePoolName = "test"
+	nodeServer.cfg.StoragePoolDataDir = make(map[string]string)
+	nodeServer.cfg.StoragePoolDataDir["test"] = tempDir
+	_, err = nodeServer.NodePublishVolume(context.TODO(), &csi.NodePublishVolumeRequest{
+		VolumeId: "csi-abcd",
+		VolumeContext: map[string]string{
+			ephemeralContextKey: "true",
+		},
+		TargetPath: filepath.Join(tempDir, validVolId),
+		VolumeCapability: &csi.VolumeCapability{
+			AccessType: &csi.VolumeCapability_Mount{
+				Mount: &csi.VolumeCapability_MountVolume{
+					FsType: "ext4",
+				},
+			},
+		},
+	})
+	Expect(err).ToNot(HaveOccurred())
+	info, err := os.Stat(filepath.Join(tempDir, "csi-abcd"))
+	Expect(err).ToNot(HaveOccurred())
+	Expect(info.IsDir()).To(BeTrue())
+	Expect(len(fakeMounter.GetLog())).To(Equal(1))
+	mountAction := fakeMounter.GetLog()[0]
+	Expect(mountAction.Action).To(Equal(mount.FakeActionMount))
+	Expect(mountAction.Target).To(Equal(filepath.Join(tempDir, validVolId)))
+	Expect(mountAction.Source).To(Equal(filepath.Join(tempDir, "csi-abcd")))
+	Expect(mountAction.FSType).To(Equal("ext4"))
+	Expect(len(fakeMounter.MountPoints)).To(Equal(1))
+	mountPoint := fakeMounter.MountPoints[0]
+	Expect(mountPoint.Opts).To(ContainElement("bind"))
+	t.Log("Unpublish ephemeral")
+	_, err = nodeServer.NodeUnpublishVolume(context.TODO(), &csi.NodeUnpublishVolumeRequest{
+		VolumeId: "csi-abcd",
+		TargetPath: filepath.Join(tempDir, validVolId),
+	})
+	Expect(err).ToNot(HaveOccurred())
+	_, err = os.Stat(filepath.Join(tempDir, "csi-abcd"))
+	Expect(err).To(HaveOccurred())
+	Expect(os.IsNotExist(err)).To(BeTrue())
+	t.Log("Unpublish ephemeral again")
+	_, err = nodeServer.NodeUnpublishVolume(context.TODO(), &csi.NodeUnpublishVolumeRequest{
+		VolumeId: "csi-abcd",
+		TargetPath: filepath.Join(tempDir, validVolId),
+	})
+	Expect(err).ToNot(HaveOccurred())
+	_, err = os.Stat(filepath.Join(tempDir, "csi-abcd"))
+	Expect(err).To(HaveOccurred())
+	Expect(os.IsNotExist(err)).To(BeTrue())
+}
+
 func Test_NodePublishVolumeValidMountPointReadOnly(t *testing.T) {
 	RegisterTestingT(t)
 	tempDir, err := ioutil.TempDir(os.TempDir(), "")
diff --git a/pkg/hostpath/utils.go b/pkg/hostpath/utils.go
@@ -16,8 +16,10 @@ limitations under the License.
 package hostpath
 
 import (
+	"io/ioutil"
 	"os"
 	"path/filepath"
+	"sort"
 
 	"k8s.io/klog/v2"
 )
@@ -105,4 +107,21 @@ func getStoragePoolNameFromMap(params map[string]string) string {
 		return params[storagePoolName]
 	}
 	return legacyStoragePoolName
-}
+}
+
+func getVolumeDirectories(storagePoolDataDirs map[string]string) ([]string, error) {
+	directories := make([]string, 0)
+	for _, path := range storagePoolDataDirs {
+		files, err := ioutil.ReadDir(path)
+		if err != nil {
+			return nil, err
+		}
+		for _, file := range files {
+			if file.IsDir() {
+				directories = append(directories, filepath.Join(path, file.Name()))
+			}
+		}
+	}
+	sort.Strings(directories)
+	return directories, nil
+}

Original file line number	Diff line number	Diff line change
`@@ -39,6 +39,7 @@ type Config struct {`
`39`	`39`	`Endpoint string`
`40`	`40`	`NodeID string`
`41`	`41`	`StoragePoolDataDir map[string]string`
	`42`	`+ DefaultStoragePoolName string`
`42`	`43`	`Version string`
`43`	`44`	`Mounter mount.Interface`
`44`	`45`	`}`
`@@ -68,13 +69,16 @@ func NewHostPathDriver(cfg Config, dataDir string) (hostPath, error) {`
`68`	`69`	`if cfg.Mounter == nil {`
`69`	`70`	`cfg.Mounter = mount.New("")`
`70`	`71`	`}`
`71`		`- cfg.StoragePoolDataDir = make(map[string]string, 0)`
	`72`	`+ cfg.StoragePoolDataDir = make(map[string]string)`
`72`	`73`
`73`	`74`	`storagePools := make([]StoragePoolInfo, 0)`
`74`	`75`	`if err := json.Unmarshal([]byte(dataDir), &storagePools); err != nil {`
`75`	`76`	`return nil, errors.New("unable to parse storage pool info")`
`76`	`77`	`}`
`77`	`78`	`for _, storagePool := range storagePools {`
	`79`	`+ if len(cfg.DefaultStoragePoolName) == 0 {`
	`80`	`+ cfg.DefaultStoragePoolName = storagePool.Name`
	`81`	`+ }`
`78`	`82`	`cfg.StoragePoolDataDir[storagePool.Name] = storagePool.Path`
`79`	`83`	`}`
`80`	`84`