feat: add glusterfs (#222)

Author: luoyuedong

pkg/apiserver/common/utils.go

@@ -52,6 +52,7 @@ const (
 	SFTP                          = "sftp"
 	Mock                          = "mock"
 	CFS                           = "cfs"
+	Glusterfs                     = "glusterfs"
 	IPDomainOrIPDomainPortPattern = "^([a-zA-Z0-9][-a-zA-Z0-9]{0,62}(\\.[a-zA-Z0-9][-a-zA-Z0-9]{0,62})+)" +
 		"(:([1-9]|[1-9]\\d{1,3}|[1-5]\\d{4}|6[0-4]\\d{3}|65[0-4]\\d{2}|655[0-2]\\d|6553[0-5]))?$"
 
@@ -160,6 +161,10 @@ func InformationFromURL(url string, properties map[string]string) (fileSystemTyp
 	case CFS:
 		serverAddress = urlSplit[ServerAddressSplit]
 		subPath = "/" + SubPathFromUrl(urlSplit, CFSSplit)
+	case Glusterfs:
+		glusterfsInfo := strings.Split(urlSplit[ServerAddressSplit], ":")
+		serverAddress = glusterfsInfo[0]
+		subPath = glusterfsInfo[1]
 	default:
 		return
 	}

pkg/apiserver/router/v1/fs.go

@@ -66,15 +66,16 @@ func (pr *PFSRouter) AddRouter(r chi.Router) {
 }
 
 var URLPrefix = map[string]bool{
-	common.HDFS:  true,
-	common.Local: true,
-	common.S3:    true,
-	common.SFTP:  true,
-	common.Mock:  true,
-	common.CFS:   true,
+	common.HDFS:      true,
+	common.Local:     true,
+	common.S3:        true,
+	common.SFTP:      true,
+	common.Mock:      true,
+	common.CFS:       true,
+	common.Glusterfs: true,
 }
 
-const FsNameMaxLen = 8
+const FsNameMaxLen = 20
 
 // createFileSystem the function that handle the create file system request
 // @Summary createFileSystem
@@ -173,7 +174,7 @@ func validateCreateFileSystem(ctx *logger.RequestContext, req *api.CreateFileSys
 		ctx.ErrorCode = common.InvalidFileSystemURL
 		return err
 	}
-	if fileSystemType == common.Mock {
+	if fileSystemType == common.Mock || fileSystemType == common.Glusterfs {
 		return nil
 	}
 	fsType, serverAddress, subPath := common.InformationFromURL(req.Url, req.Properties)
@@ -345,6 +346,9 @@ func checkFsDir(fsType, url string, properties map[string]string) error {
 		inputIPs = strings.Split(properties[fsCommon.Endpoint], ",")
 		subPath = "/" + strings.SplitAfterN(url, "/", 4)[3]
 	}
+	if len(inputIPs) == 0 {
+		return nil
+	}
 	fsList, err := models.GetSimilarityAddressList(fsType, inputIPs)
 	if err != nil {
 		return err

pkg/fs/csiplugin/client/pfs/mount_info.go

@@ -20,10 +20,12 @@ import (
 	"fmt"
 	"strings"
 
+	"paddleflow/pkg/apiserver/common"
 	csiCommon "paddleflow/pkg/fs/utils/common"
 )
 
 const (
+	mountName       = "mount"
 	pfsMountCmdName = "./mount.sh"
 	ReadOnly        = "ro"
 	ReadWrite       = "rw"
@@ -44,31 +46,44 @@ type FSMountParameter struct {
 }
 
 type MountInfo struct {
-	Server       string
-	FSID         string
-	TargetPath   string
-	LocalPath    string
-	UsernameRoot string
-	PasswordRoot string
-	UID          int
-	GID          int
-	Options      []string
+	Server        string
+	FSID          string
+	TargetPath    string
+	LocalPath     string
+	UsernameRoot  string
+	PasswordRoot  string
+	ClusterID     string
+	Type          string
+	ServerAddress string
+	SubPath       string
+	UID           int
+	GID           int
+	Options       []string
 }
 
 func (m *MountInfo) GetMountCmd() (string, []string) {
-	cmdName := pfsMountCmdName
+	var cmdName string
 	var args []string
-
-	if len(m.Options) > 0 {
-		mountOptions := strings.Join(m.Options[:], ",")
-		args = append(args, fmt.Sprintf(MountOptions, mountOptions))
+	if m.Type == common.Glusterfs {
+		cmdName = mountName
+		args = append(args, "-t", m.Type)
+		if len(m.Options) != 0 {
+			args = append(args, "-o", strings.Join(m.Options, ","))
+		}
+		args = append(args, strings.Join([]string{m.ServerAddress, m.SubPath}, ":"), m.LocalPath)
+	} else {
+		cmdName = pfsMountCmdName
+		if len(m.Options) > 0 {
+			mountOptions := strings.Join(m.Options[:], ",")
+			args = append(args, fmt.Sprintf(MountOptions, mountOptions))
+		}
+		// set pfs-fuse arguments
+		args = append(args, fmt.Sprintf(PFSServerOption, m.Server))
+		args = append(args, fmt.Sprintf(FSIDOptions, m.FSID))
+		args = append(args, fmt.Sprintf(UIDOption, m.UID))
+		args = append(args, fmt.Sprintf(GIDOption, m.GID))
+		args = append(args, fmt.Sprintf(MountPoint, m.LocalPath))
 	}
-	// set pfs-fuse arguments
-	args = append(args, fmt.Sprintf(PFSServerOption, m.Server))
-	args = append(args, fmt.Sprintf(FSIDOptions, m.FSID))
-	args = append(args, fmt.Sprintf(UIDOption, m.UID))
-	args = append(args, fmt.Sprintf(GIDOption, m.GID))
-	args = append(args, fmt.Sprintf(MountPoint, m.LocalPath))
 
 	return cmdName, args
 }

pkg/fs/csiplugin/csidriver/node_server.go

@@ -27,10 +27,8 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
-	"paddleflow/pkg/client"
-	"paddleflow/pkg/common/http/api"
-	"paddleflow/pkg/fs/client/base"
 	"paddleflow/pkg/fs/csiplugin/client/pfs"
+	"paddleflow/pkg/fs/csiplugin/mount"
 	"paddleflow/pkg/fs/utils/common"
 	"paddleflow/pkg/fs/utils/io"
 	mountUtil "paddleflow/pkg/fs/utils/mount"
@@ -76,28 +74,10 @@ func (ns *nodeServer) NodePublishVolume(ctx context.Context,
 	fsId := volumeContext[pfsFSID]
 	server := volumeContext[pfsServer]
 
-	// new fuse http client
-	httpClient := client.NewHttpClient(server, client.DefaultTimeOut)
-	// token
-	login := api.LoginParams{
-		UserName: ns.credentialInfo.usernameRoot,
-		Password: ns.credentialInfo.passwordRoot,
-	}
-	loginResponse, err := api.LoginRequest(login, httpClient)
-	if err != nil {
-		log.Errorf("fuse login failed: %v", err)
-		return &csi.NodePublishVolumeResponse{}, err
-	}
-	_, err = base.NewClient(fsId, httpClient, loginResponse.Authorization)
-	if err != nil {
-		log.Errorf("csi addRefOfMount: init client with fs[%s] and server[%s] failed: %v",
-			fsId, server, err)
-		return &csi.NodePublishVolumeResponse{}, err
-	}
-
 	mountInfo := pfs.GetMountInfo(fsId, server, req.GetReadonly())
+	mountInfo.UsernameRoot, mountInfo.PasswordRoot = ns.credentialInfo.usernameRoot, ns.credentialInfo.passwordRoot
+	mountInfo.TargetPath = targetPath
 	// root credentials for pfs-fuse
-	mountInfo.UsernameRoot, mountInfo.PasswordRoot = login.UserName, login.Password
 	pathPrefix := filepath.Dir(targetPath)
 	mountInfo.TargetPath = targetPath
 	if err := mountVolume(pathPrefix, mountInfo, req.GetReadonly()); err != nil {
@@ -147,6 +127,11 @@ func (ns *nodeServer) NodeExpandVolume(ctx context.Context,
 
 func mountVolume(mountPathPrefix string, mountInfo pfs.MountInfo, readOnly bool) error {
 	log.Debugf("mountVolume mountInfo:%+v, readOnly:%t", mountInfo, readOnly)
+	_, _, err := mount.HttpClientAndToken(&mountInfo)
+	if err != nil {
+		log.Errorf("GetFsInfoAndClient err: %v", err)
+		return err
+	}
 	// business pods use a separate source path
 	volumeSourceMountPath := common.GetVolumeSourceMountPath(mountPathPrefix)
 	if err := os.MkdirAll(volumeSourceMountPath, 0750); err != nil {
@@ -162,17 +147,8 @@ func mountVolume(mountPathPrefix string, mountInfo pfs.MountInfo, readOnly bool)
 		log.Errorf("exec mount failed: [%v], output[%v]", err, string(output))
 		return err
 	}
-
-	//err := mount.MountThroughPod(mountInfo)
-	//if err != nil {
-	//	log.Errorf("MountThroughPod err: %v", err)
-	//	return err
-	//}
-
 	volumeBindMountPath := common.GetVolumeMountPath(mountPathPrefix)
 	return bindMountVolume(volumeSourceMountPath, volumeBindMountPath, readOnly)
-	//bindSource := mount.MountDir + "/" + mountInfo.FSID + "/storage"
-	//return bindMountVolume(bindSource, mountInfo.TargetPath, readOnly)
 }
 
 func bindMountVolume(sourcePath, mountPath string, readOnly bool) error {

pkg/fs/csiplugin/mount/pod.go

@@ -30,7 +30,9 @@ import (
 	v1 "k8s.io/api/core/v1"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 
+	"paddleflow/pkg/client"
 	"paddleflow/pkg/common/http/api"
+	"paddleflow/pkg/common/http/core"
 	"paddleflow/pkg/fs/client/base"
 	"paddleflow/pkg/fs/common"
 	"paddleflow/pkg/fs/csiplugin/client/k8s"
@@ -274,3 +276,50 @@ func getcmd(mountInfo pfs.MountInfo, cacheConf common.FsCacheConfig) string {
 	cmd := mkdir + pfsMountPath + mountPath + strings.Join(options, " ")
 	return cmd
 }
+
+func HttpClientAndToken(mountInfo *pfs.MountInfo) (*core.PFClient, string, error) {
+	// login server
+	httpClient := client.NewHttpClient(mountInfo.Server, client.DefaultTimeOut)
+	login := api.LoginParams{
+		UserName: mountInfo.UsernameRoot,
+		Password: mountInfo.PasswordRoot,
+	}
+	loginResponse, err := api.LoginRequest(login, httpClient)
+	if err != nil {
+		log.Errorf("Mount: login failed: %v", err)
+		return nil, "", err
+	}
+	token := loginResponse.Authorization
+	// validate fs
+	var resp *api.FsResponse
+	if resp, err = getFs(mountInfo.FSID, httpClient, token); err != nil {
+		log.Errorf("Mount: validate fs exist [%s] err: %v", mountInfo.FSID, err)
+		return nil, "", err
+	}
+	mountInfo.Type = resp.Type
+	mountInfo.SubPath = resp.SubPath
+	mountInfo.ServerAddress = resp.ServerAddress
+	return httpClient, token, nil
+}
+
+func getFs(fsID string, httpClient *core.PFClient, token string) (*api.FsResponse, error) {
+	userName, fsName := getFsNameAndUserNameByFsID(fsID)
+	params := api.FsParams{
+		FsName:   fsName,
+		UserName: userName,
+		Token:    token,
+	}
+	fsResp, err := api.FsRequest(params, httpClient)
+	if err != nil {
+		log.Errorf("fs request[%+v] failed: %v", params, err)
+		return nil, err
+	}
+	return fsResp, nil
+}
+
+func getFsNameAndUserNameByFsID(fsID string) (userName string, fsName string) {
+	fsArray := strings.Split(fsID, "-")
+	userName = strings.Join(fsArray[1:len(fsArray)-1], "")
+	fsName = fsArray[len(fsArray)-1]
+	return userName, fsName
+}