fuse: Integrate fdstore into cfs-client

This patch integrate fdstore into cfs-client, and adds 2 more options
for cfs-client:
 -p <port>: pprof port of old cfs-client
 -n: dynamic unix domain socket file name

If we are trying to live upgrade cfs-client, the following command
could be used:

	cfs-client -c <config json file path> -p <pprof port> -r

Signed-off-by: Sheng Yong <shengyong2021@gmail.com>

client/fuse.go

@@ -23,6 +23,7 @@ package main
 import (
 	"flag"
 	"fmt"
+	"io/ioutil"
 	syslog "log"
 	"net"
 	"net/http"
@@ -38,6 +39,7 @@ import (
 	"time"
 
 	"github.com/chubaofs/chubaofs/sdk/master"
+	"github.com/chubaofs/chubaofs/util"
 
 	sysutil "github.com/chubaofs/chubaofs/util/sys"
 
@@ -74,15 +76,19 @@ const (
 	ControlCommandResume       = "/resume"
 	Role                       = "Client"
 
-	DefaultFuseUDS = "/tmp/ChubaoFS-fdstore.sock"
+	DefaultIP            = "127.0.0.1"
+	DynamicUDSNameFormat = "/tmp/ChubaoFS-fdstore-%v.sock"
+	DefaultUDSName       = "/tmp/ChubaoFS-fdstore.sock"
 )
 
 var (
 	configFile           = flag.String("c", "", "FUSE client config file")
 	configVersion        = flag.Bool("v", false, "show version")
 	configForeground     = flag.Bool("f", false, "run foreground")
+	configDynamicUDSName = flag.Bool("n", false, "dynamic unix domain socket filename")
 	configRestoreFuse    = flag.Bool("r", false, "restore FUSE instead of mounting")
 	configRestoreFuseUDS = flag.String("s", "", "restore socket addr")
+	configFuseHttpPort   = flag.String("p", "", "fuse http service port")
 )
 
 var GlobalMountOptions []proto.MountOption
@@ -92,6 +98,155 @@ func init() {
 	proto.InitMountOptions(GlobalMountOptions)
 }
 
+func createUDS(sockAddr string) (listener net.Listener, err error) {
+	var addr *net.UnixAddr
+
+	log.LogInfof("sockaddr: %s\n", sockAddr)
+
+	os.Remove(sockAddr)
+	if addr, err = net.ResolveUnixAddr("unix", sockAddr); err != nil {
+		log.LogErrorf("cannot resolve unix addr: %v\n", err)
+		return
+	}
+
+	if listener, err = net.ListenUnix("unix", addr); err != nil {
+		log.LogErrorf("cannot create unix domain: %v\n", err)
+		return
+	}
+
+	if err = os.Chmod(sockAddr, 0666); err != nil {
+		log.LogErrorf("failed to chmod socket file: %v\n", err)
+		listener.Close()
+		return
+	}
+
+	return
+}
+
+func destroyUDS(listener net.Listener) {
+	sockAddr := listener.Addr().String()
+	listener.Close()
+	os.Remove(sockAddr)
+}
+
+func recvFuseFdFromOldClient(udsListener net.Listener) (file *os.File, err error) {
+	var conn net.Conn
+	var socket *os.File
+
+	if conn, err = udsListener.Accept(); err != nil {
+		log.LogErrorf("unix domain accepts fail: %v\n", err)
+		return
+	}
+	defer conn.Close()
+
+	log.LogInfof("a new connection accepted\n")
+	unixconn := conn.(*net.UnixConn)
+	if socket, err = unixconn.File(); err != nil {
+		log.LogErrorf("failed to get socket file: %v\n", err)
+		return
+	}
+	defer socket.Close()
+
+	if file, err = util.RecvFd(socket); err != nil {
+		log.LogErrorf("failed to receive fd: %v\n", err)
+		return
+	}
+
+	log.LogInfof("Received file %s fd %v\n", file.Name(), file.Fd())
+	return
+}
+
+func sendSuspendRequest(port string, udsListener net.Listener) (err error) {
+	var (
+		req  *http.Request
+		resp *http.Response
+		data []byte
+	)
+	udsFilePath := udsListener.Addr().String()
+
+	url := fmt.Sprintf("http://%s:%s/suspend?sock=%s", DefaultIP, port, udsFilePath)
+	if req, err = http.NewRequest("POST", url, nil); err != nil {
+		log.LogErrorf("Failed to get new request: %v\n", err)
+		return err
+	}
+	req.Header.Set("Content-Type", "application/text")
+
+	client := http.DefaultClient
+	client.Timeout = 120 * time.Second
+	if resp, err = client.Do(req); err != nil {
+		log.LogErrorf("Failed to post request: %v\n", err)
+		return err
+	}
+	defer resp.Body.Close()
+
+	if data, err = ioutil.ReadAll(resp.Body); err != nil {
+		log.LogErrorf("Failed to read response: %v\n", err)
+		return err
+	}
+
+	if resp.StatusCode == http.StatusOK {
+		log.LogInfof("\n==> %s\n==> Could restore cfs-client now with -r option.\n\n", string(data))
+	} else {
+		log.LogErrorf("\n==> %s\n==> Status: %s\n\n", string(data), resp.Status)
+		return fmt.Errorf(resp.Status)
+	}
+
+	return nil
+}
+
+func sendResumeRequest(port string) (err error) {
+	var (
+		req  *http.Request
+		resp *http.Response
+		data []byte
+	)
+
+	url := fmt.Sprintf("http://%s:%s/resume", DefaultIP, port)
+	if req, err = http.NewRequest("POST", url, nil); err != nil {
+		log.LogErrorf("Failed to get new request: %v\n", err)
+		return err
+	}
+	req.Header.Set("Content-Type", "application/text")
+
+	client := http.DefaultClient
+	if resp, err = client.Do(req); err != nil {
+		log.LogErrorf("Failed to post request: %v\n", err)
+		return err
+	}
+	defer resp.Body.Close()
+
+	if data, err = ioutil.ReadAll(resp.Body); err != nil {
+		log.LogErrorf("Failed to read response: %v\n", err)
+		return err
+	}
+
+	log.LogInfof("data: %s\n", string(data))
+	return nil
+}
+
+func doSuspend(uds string, port string) (*os.File, error) {
+	var fud *os.File
+
+	udsListener, err := createUDS(uds)
+	if err != nil {
+		log.LogErrorf("doSuspend: failed to create UDS: %v\n", err)
+		return nil, err
+	}
+	defer destroyUDS(udsListener)
+
+	if err = sendSuspendRequest(port, udsListener); err != nil {
+		sendResumeRequest(port)
+		return nil, err
+	}
+
+	if fud, err = recvFuseFdFromOldClient(udsListener); err != nil {
+		sendResumeRequest(port)
+		return nil, err
+	}
+
+	return fud, nil
+}
+
 func main() {
 	flag.Parse()
 
@@ -183,6 +338,27 @@ func main() {
 		os.Exit(1)
 	}
 
+	var fud *os.File
+	if opt.NeedRestoreFuse && *configFuseHttpPort != "" {
+		log.LogInfof("Suspend/Restore by self\n")
+		var udsName string
+		if *configDynamicUDSName {
+			udsName = fmt.Sprintf(DynamicUDSNameFormat, os.Getpid())
+		} else {
+			udsName = DefaultUDSName
+		}
+
+		// Tell old cfs-client to suspend first. This should be done
+		// before mount() to avoid pprof port conflict between old and
+		// new cfs-clients.
+		if fud, err = doSuspend(udsName, *configFuseHttpPort); err != nil {
+			log.LogErrorf("Failed to tell old cfs-client to suspend: %v\n", err)
+			syslog.Printf("Error: Failed to tell old cfs-client to suspend: %v\n", err)
+			log.LogFlush()
+			os.Exit(1)
+		}
+	}
+
 	fsConn, super, err := mount(opt)
 	if err != nil {
 		err = errors.NewErrorf("mount failed: %v", err)
@@ -206,10 +382,14 @@ func main() {
 	}
 
 	if opt.NeedRestoreFuse {
-		if *configRestoreFuseUDS == "" {
-			super.SetSockAddr(DefaultFuseUDS)
+		if fud == nil {
+			if *configRestoreFuseUDS == "" {
+				super.SetSockAddr(DefaultUDSName)
+			} else {
+				super.SetSockAddr(*configRestoreFuseUDS)
+			}
 		} else {
-			super.SetSockAddr(*configRestoreFuseUDS)
+			fsConn.SetFuseDevFile(fud)
 		}
 	}
 

vendor/bazil.org/fuse/fs/serve.go

@@ -718,8 +718,10 @@ func (s *Server) TryRestore(fs FS) error {
 	if err != nil {
 		return err
 	}
-	if err = s.LoadFuseDevFd(sockaddr); err != nil {
-		return err
+	if s.conn.GetFuseDevFile() == nil {
+		if err = s.LoadFuseDevFd(sockaddr); err != nil {
+			return err
+		}
 	}
 
 	fs.Notify(stat, "")