Merge pull request #55866 from zhangxiaoyu-zidif/make-k8s-support-ceph-fuse

Automatic merge from submit-queue (batch tested with PRs 55954, 56037, 55866, 55984, 54994). If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>.

make k8s support cephfs fuse mount

**What this PR does / why we need it**:
make  k8s support cephfs fuse.

**Which issue(s) this PR fixes** *(optional, in `fixes #<issue number>(, fixes #<issue_number>, ...)` format, will close the issue(s) when PR gets merged)*:
Fixes https://github.com/kubernetes/kubernetes/issues/46636, https://github.com/kubernetes/kubernetes/issues/44486

**Special notes for your reviewer**:

**Release note**:

```release-note
K8s supports cephfs fuse mount.
```

pkg/volume/cephfs/cephfs.go

@@ -19,6 +19,9 @@ package cephfs
 import (
 	"fmt"
 	"os"
+	"os/exec"
+	"path"
+	"runtime"
 	"strings"
 
 	"github.com/golang/glog"
@@ -231,15 +234,32 @@ func (cephfsVolume *cephfsMounter) SetUpAt(dir string, fsGroup *int64) error {
 	}
 	os.MkdirAll(dir, 0750)
 
-	err = cephfsVolume.execMount(dir)
-	if err == nil {
-		return nil
+	// check whether it belongs to fuse, if not, default to use kernel mount.
+	if cephfsVolume.checkFuseMount() {
+		glog.V(4).Infof("CephFS fuse mount.")
+		err = cephfsVolume.execFuseMount(dir)
+		// cleanup no matter if fuse mount fail.
+		keyringPath := cephfsVolume.GetKeyringPath()
+		_, StatErr := os.Stat(keyringPath)
+		if !os.IsNotExist(StatErr) {
+			os.RemoveAll(keyringPath)
+		}
+		if err == nil {
+			// cephfs fuse mount succeeded.
+			return nil
+		} else {
+			// if cephfs fuse mount failed, fallback to kernel mount.
+			glog.V(4).Infof("CephFS fuse mount failed: %v, fallback to kernel mount.", err)
+		}
 	}
-
-	// cleanup upon failure
-	util.UnmountPath(dir, cephfsVolume.mounter)
-	// return error
-	return err
+	glog.V(4).Infof("CephFS kernel mount.")
+	err = cephfsVolume.execMount(dir)
+	if err != nil {
+		// cleanup upon failure.
+		util.UnmountPath(dir, cephfsVolume.mounter)
+		return err
+	}
+	return nil
 }
 
 type cephfsUnmounter struct {
@@ -264,6 +284,14 @@ func (cephfsVolume *cephfs) GetPath() string {
 	return cephfsVolume.plugin.host.GetPodVolumeDir(cephfsVolume.podUID, utilstrings.EscapeQualifiedNameForDisk(name), cephfsVolume.volName)
 }
 
+// GetKeyringPath creates cephfuse keyring path
+func (cephfsVolume *cephfs) GetKeyringPath() string {
+	name := cephfsPluginName
+	volumeDir := cephfsVolume.plugin.host.GetPodVolumeDir(cephfsVolume.podUID, utilstrings.EscapeQualifiedNameForDisk(name), cephfsVolume.volName)
+	volumeKeyringDir := volumeDir + "~keyring"
+	return volumeKeyringDir
+}
+
 func (cephfsVolume *cephfs) execMount(mountpoint string) error {
 	// cephfs mount option
 	ceph_opt := ""
@@ -299,6 +327,92 @@ func (cephfsVolume *cephfs) execMount(mountpoint string) error {
 	return nil
 }
 
+func (cephfsMounter *cephfsMounter) checkFuseMount() bool {
+	execute := cephfsMounter.plugin.host.GetExec(cephfsMounter.plugin.GetPluginName())
+	switch runtime.GOOS {
+	case "linux":
+		retBytes, err := execute.Run("/bin/ls", "/sbin/mount.fuse.ceph")
+		if err == nil && string(retBytes) == "/sbin/mount.fuse.ceph\n" {
+			glog.V(4).Infof("/sbin/mount.fuse.ceph exists, it should be fuse mount")
+			return true
+		}
+		return false
+	}
+	return false
+}
+
+func (cephfsVolume *cephfs) execFuseMount(mountpoint string) error {
+	// cephfs keyring file
+	keyring_file := ""
+	// override secretfile if secret is provided
+	if cephfsVolume.secret != "" {
+		// TODO: cephfs fuse currently doesn't support secret option,
+		// remove keyring file create once secret option is supported.
+		glog.V(4).Infof("cephfs mount begin using fuse.")
+
+		keyringPath := cephfsVolume.GetKeyringPath()
+		os.MkdirAll(keyringPath, 0750)
+
+		payload := make(map[string]util.FileProjection, 1)
+		var fileProjection util.FileProjection
+
+		keyring := fmt.Sprintf("[client.%s]\n", cephfsVolume.id) + "key = " + cephfsVolume.secret + "\n"
+
+		fileProjection.Data = []byte(keyring)
+		fileProjection.Mode = int32(0644)
+		fileName := cephfsVolume.id + ".keyring"
+
+		payload[fileName] = fileProjection
+
+		writerContext := fmt.Sprintf("cephfuse:%v.keyring", cephfsVolume.id)
+		writer, err := util.NewAtomicWriter(keyringPath, writerContext)
+		if err != nil {
+			glog.Errorf("failed to create atomic writer: %v", err)
+			return err
+		}
+
+		err = writer.Write(payload)
+		if err != nil {
+			glog.Errorf("failed to write payload to dir: %v", err)
+			return err
+		}
+
+		keyring_file = path.Join(keyringPath, fileName)
+
+	} else {
+		keyring_file = cephfsVolume.secret_file
+	}
+
+	// build src like mon1:6789,mon2:6789,mon3:6789:/
+	hosts := cephfsVolume.mon
+	l := len(hosts)
+	// pass all monitors and let ceph randomize and fail over
+	i := 0
+	src := ""
+	for i = 0; i < l-1; i++ {
+		src += hosts[i] + ","
+	}
+	src += hosts[i]
+
+	mountArgs := []string{}
+	mountArgs = append(mountArgs, "-k")
+	mountArgs = append(mountArgs, keyring_file)
+	mountArgs = append(mountArgs, "-m")
+	mountArgs = append(mountArgs, src)
+	mountArgs = append(mountArgs, mountpoint)
+	mountArgs = append(mountArgs, "-r")
+	mountArgs = append(mountArgs, cephfsVolume.path)
+
+	glog.V(4).Infof("Mounting cmd ceph-fuse with arguments (%s)", mountArgs)
+	command := exec.Command("ceph-fuse", mountArgs...)
+	output, err := command.CombinedOutput()
+	if err != nil || !(strings.Contains(string(output), "starting fuse")) {
+		return fmt.Errorf("Ceph-fuse failed: %v\narguments: %s\nOutput: %s\n", err, mountArgs, string(output))
+	}
+
+	return nil
+}
+
 func getVolumeSource(spec *volume.Spec) ([]string, string, string, string, bool, error) {
 	if spec.Volume != nil && spec.Volume.CephFS != nil {
 		mon := spec.Volume.CephFS.Monitors