diff --git a/cmd/fs/fuse/flag/flags.go b/cmd/fs/fuse/flag/flags.go index f4c576db..2868667f 100644 --- a/cmd/fs/fuse/flag/flags.go +++ b/cmd/fs/fuse/flag/flags.go @@ -181,6 +181,12 @@ func MountFlags(fuseConf *meta.FuseConfig) []cli.Flag { Value: false, Usage: "Assume all directory objects (\"dir/\") exist (default: false)", }, + &cli.BoolFlag{ + Name: "background", + Aliases: []string{"d"}, + Value: false, + Usage: "run in backGround", + }, } } diff --git a/cmd/fs/fuse/service/mount.go b/cmd/fs/fuse/service/mount.go index ad9669fb..bbc40948 100644 --- a/cmd/fs/fuse/service/mount.go +++ b/cmd/fs/fuse/service/mount.go @@ -31,6 +31,7 @@ import ( "syscall" "time" + utils2 "github.com/PaddlePaddle/PaddleFlow/pkg/fs/client/utils" libfuse "github.com/hanwen/go-fuse/v2/fuse" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promhttp" @@ -234,6 +235,10 @@ func mount(c *cli.Context) error { return err } + if c.Bool("background") { + daemonRun() + } + if !c.Bool("local") { stopChan := make(chan struct{}) defer close(stopChan) @@ -248,7 +253,7 @@ func mount(c *cli.Context) error { } } - log.Debugf("start mount service") + log.Infof("start mount service on %v", c.String("mount-point")) server, err := fuse.Server(c.String("mount-point"), *opts) if err != nil { log.Fatalf("mount fail: %v", err) @@ -279,6 +284,9 @@ func InitVFS(c *cli.Context, registry *prometheus.Registry) error { server := c.String("server") if c.Bool("local") == true { localRoot := c.String("local-root") + if localRoot == "" { + localRoot = "./mock" + } if localRoot == "" || localRoot == "/" { log.Errorf("invalid localRoot: [%s]", localRoot) return fmt.Errorf("invalid localRoot: [%s]", localRoot) @@ -420,6 +428,9 @@ func InitVFS(c *cli.Context, registry *prometheus.Registry) error { CachePath: c.String("data-cache-path"), }, } + if fsMeta.Properties == nil { + fsMeta.Properties = make(map[string]string) + } if c.Bool("no-implicit-dir") { fsMeta.Properties[common.ImplicitDir] = "false" } else { @@ -460,3 +471,19 @@ func signalHandle(mp string) { } }() } + +func daemonRun() { + err := makeDaemon() + if err != nil { + log.Fatalf("Failed to make daemon: %s", err) + } + if runtime.GOOS == "linux" { + log.SetOutput(os.Stderr) + } +} + +func makeDaemon() error { + var attrs utils2.DaemonAttr + _, _, err := utils2.MakeDaemon(&attrs) + return err +} diff --git a/cmd/fs/fuse/service/mount_test.go b/cmd/fs/fuse/service/mount_test.go new file mode 100644 index 00000000..ecb9e2bf --- /dev/null +++ b/cmd/fs/fuse/service/mount_test.go @@ -0,0 +1,65 @@ +package service + +import ( + "github.com/PaddlePaddle/PaddleFlow/pkg/fs/utils" + "github.com/PaddlePaddle/PaddleFlow/pkg/version" + "github.com/urfave/cli/v2" + "os" + "testing" + "time" +) + +const testMountPoint = "./mock" + +func MockMain(args []string) error { + cli.VersionFlag = &cli.BoolFlag{ + Name: "version", Aliases: []string{"V"}, + Usage: "print only the version", + } + app := &cli.App{ + Name: "pfs-fuse", + Usage: "A POSIX file system built on kv DB and object storage.", + Version: version.InfoStr(), + Copyright: "Apache License 2.0", + HideHelpCommand: true, + EnableBashCompletion: true, + Commands: []*cli.Command{ + CmdMount(), + CmdUmount(), + }, + } + return app.Run(args) +} + +func mountTemp(t *testing.T) { + os.MkdirAll(testMountPoint, 0755) + mountArgs := []string{"", "mount", "-mp", testMountPoint, "--local", "true", "--local-root", testMountPoint, "-d", "true"} + + go func() { + if err := MockMain(mountArgs); err != nil { + t.Errorf("mount failed: %s", err) + } + }() + time.Sleep(3 * time.Second) + + inode, err := utils.GetFileInode(testMountPoint) + if err != nil { + t.Fatalf("get file inode failed: %s", err) + } + if inode != 1 { + t.Fatalf("mount failed: inode of %s is not 1", testMountPoint) + } else { + t.Logf("mount %s success", testMountPoint) + } +} + +func umountTemp(t *testing.T) { + if err := MockMain([]string{"", "umount", testMountPoint}); err != nil { + t.Fatalf("umount failed: %s", err) + } +} + +func TestMountBackground(t *testing.T) { + mountTemp(t) + defer umountTemp(t) +} diff --git a/go.mod b/go.mod index 7f8d33b1..238db105 100644 --- a/go.mod +++ b/go.mod @@ -20,7 +20,7 @@ require ( github.com/go-chi/chi v1.5.4 github.com/google/uuid v1.3.0 github.com/gorilla/websocket v1.4.2 - github.com/hanwen/go-fuse/v2 v2.2.0 + github.com/hanwen/go-fuse/v2 v2.5.1 github.com/jcmturner/gokrb5/v8 v8.4.4 github.com/jinzhu/copier v0.3.2 github.com/johannesboyne/gofakes3 v0.0.0-20220627085814-c3ac35da23b2 @@ -147,7 +147,7 @@ require ( go.opencensus.io v0.22.5 // indirect golang.org/x/mod v0.8.0 // indirect golang.org/x/oauth2 v0.7.0 // indirect - golang.org/x/sys v0.15.0 // indirect + golang.org/x/sys v0.19.0 // indirect golang.org/x/term v0.15.0 // indirect golang.org/x/text v0.14.0 // indirect golang.org/x/time v0.0.0-20210723032227-1f47c861a9ac // indirect diff --git a/go.sum b/go.sum index 18cce105..1749b297 100644 --- a/go.sum +++ b/go.sum @@ -468,6 +468,8 @@ github.com/grpc-ecosystem/grpc-gateway v1.14.6/go.mod h1:zdiPV4Yse/1gnckTHtghG4G github.com/grpc-ecosystem/grpc-gateway v1.16.0/go.mod h1:BDjrQk3hbvj6Nolgz8mAMFbcEtjT1g+wF4CSlocrBnw= github.com/hanwen/go-fuse/v2 v2.2.0 h1:jo5QZYmBLNcl9ovypWaQ5yXMSSV+Ch68xoC3rtZvvBM= github.com/hanwen/go-fuse/v2 v2.2.0/go.mod h1:B1nGE/6RBFyBRC1RRnf23UpwCdyJ31eukw34oAKukAc= +github.com/hanwen/go-fuse/v2 v2.5.1 h1:OQBE8zVemSocRxA4OaFJbjJ5hlpCmIWbGr7r0M4uoQQ= +github.com/hanwen/go-fuse/v2 v2.5.1/go.mod h1:xKwi1cF7nXAOBCXujD5ie0ZKsxc8GGSA1rlMJc+8IJs= github.com/hashicorp/consul/api v1.3.0/go.mod h1:MmDNSzIMUjNpY/mQ398R4bk2FnqQLoPndWW5VkKPlCE= github.com/hashicorp/consul/sdk v0.3.0/go.mod h1:VKf9jXwCTEY1QZP2MOLRhb5i/I/ssyNV1vwHyQBF0x8= github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= @@ -716,6 +718,7 @@ github.com/mitchellh/mapstructure v0.0.0-20160808181253-ca63d7c062ee/go.mod h1:F github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y= github.com/mitchellh/mapstructure v1.4.1 h1:CpVNEelQCZBooIPDn+AR3NpivK/TIKU8bDxdASFVQag= github.com/mitchellh/mapstructure v1.4.1/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo= +github.com/moby/sys/mountinfo v0.6.2/go.mod h1:IJb6JQeOklcdMU9F5xQ8ZALD+CUr5VlGpwtX+VE0rpI= github.com/moby/term v0.0.0-20200312100748-672ec06f55cd h1:aY7OQNf2XqY/JQ6qREWamhI/81os/agb2BAGpcx5yWI= github.com/moby/term v0.0.0-20200312100748-672ec06f55cd/go.mod h1:DdlQx2hp0Ss5/fLikoLlEeIYiATotOjgB//nb973jeo= github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= @@ -1242,6 +1245,8 @@ golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.15.0 h1:h48lPFYpsTvQJZF4EKyI4aLHaev3CxivZmv7yZig9pc= golang.org/x/sys v0.15.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.19.0 h1:q5f1RH2jigJ1MoAWp2KTp3gm5zAGFUTarQZ5U386+4o= +golang.org/x/sys v0.19.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= diff --git a/pkg/fs/client/utils/daemon.go b/pkg/fs/client/utils/daemon.go new file mode 100644 index 00000000..7ca770c1 --- /dev/null +++ b/pkg/fs/client/utils/daemon.go @@ -0,0 +1,279 @@ +package utils + +import ( + "bytes" + "crypto/sha1" + "encoding/hex" + "fmt" + "io" + "os" + "strconv" + "strings" + "syscall" + "time" +) + +const ( + stageVar = "__DAEMON_STAGE" + fdVarPrefix = "__DAEMON_FD_" +) + +// DaemonAttr describes the options that apply to daemonization +type DaemonAttr struct { + ProgramName string // child's os.Args[0]; copied from parent if empty + CaptureOutput bool // whether to capture stdout/stderr + Files []**os.File // files to keep open in the daemon + Stdout *os.File // redirect stdout/stderr to it + OnExit func(stage int) error +} + +func MakeDaemon(attrs *DaemonAttr) (io.Reader, io.Reader, error) { + stage, advanceStage, resetEnv := getStage() + + fatal := func(err error) (io.Reader, io.Reader, error) { + if stage > 0 { + os.Exit(1) + } + resetErr := resetEnv() + if err != nil { + return nil, nil, resetErr + } + return nil, nil, err + } + + fileCount := 3 + len(attrs.Files) + files := make([]*os.File, fileCount, fileCount+2) + + if stage == 0 { + nullDev, err := os.OpenFile("/dev/null", 0, 0) + if err != nil { + return fatal(err) + } + files[0] = nullDev + if attrs.Stdout != nil { + files[1], files[2] = attrs.Stdout, attrs.Stdout + } else { + files[1], files[2] = nullDev, nullDev + } + + fd := 3 + for _, fPtr := range attrs.Files { + files[fd] = *fPtr + saveFileName(fd, (*fPtr).Name()) + fd++ + } + } else { + files[0], files[1], files[2] = os.Stdin, os.Stdout, os.Stderr + + fd := 3 + for _, fPtr := range attrs.Files { + *fPtr = os.NewFile(uintptr(fd), getFileName(fd)) + syscall.CloseOnExec(fd) + files[fd] = *fPtr + fd++ + } + } + + if stage < 2 { + procName, err := os.Executable() + if err != nil { + return fatal(fmt.Errorf("can't determine full path to executable: %s", err)) + } + + if len(procName) == 0 { + return fatal(fmt.Errorf("can't determine full path to executable")) + } + + if stage == 1 && attrs.CaptureOutput { + files = files[:fileCount+2] + + // stdout: write at fd:1, read at fd:fileCount + if files[fileCount], files[1], err = os.Pipe(); err != nil { + return fatal(err) + } + // stderr: write at fd:2, read at fd:fileCount+1 + if files[fileCount+1], files[2], err = os.Pipe(); err != nil { + return fatal(err) + } + } + + if err := advanceStage(); err != nil { + return fatal(err) + } + dir, _ := os.Getwd() + osAttrs := os.ProcAttr{Dir: dir, Env: os.Environ(), Files: files} + + if stage == 0 { + sysattrs := syscall.SysProcAttr{Setsid: true} + osAttrs.Sys = &sysattrs + } + + progName := attrs.ProgramName + if len(progName) == 0 { + progName = os.Args[0] + } + args := append([]string{progName}, os.Args[1:]...) + proc, err := os.StartProcess(procName, args, &osAttrs) + if err != nil { + return fatal(fmt.Errorf("can't create process %s: %s", procName, err)) + } + err = proc.Release() + if err != nil { + return nil, nil, err + } + if attrs.OnExit != nil { + err := attrs.OnExit(stage) + if err != nil { + return nil, nil, err + } + } + os.Exit(0) + } + + //os.Chdir("/") + syscall.Umask(0) + err := resetEnv() + if err != nil { + return nil, nil, err + } + + for fd := 3; fd < fileCount; fd++ { + resetFileName(fd) + } + currStage = DaemonStage(stage) + + var stdout, stderr *os.File + if attrs.CaptureOutput { + stdout = os.NewFile(uintptr(fileCount), "stdout") + stderr = os.NewFile(uintptr(fileCount+1), "stderr") + } + return stdout, stderr, nil +} + +func saveFileName(fd int, name string) { + // We encode in hex to avoid issues with filename encoding, and to be able + // to separate it from the original variable value (if set) that we want to + // keep. Otherwise, all non-zero characters are valid in the name, and we + // can't insert a zero in the var as a separator. + fdVar := fdVarPrefix + fmt.Sprint(fd) + value := fmt.Sprintf("%s:%s", + hex.EncodeToString([]byte(name)), os.Getenv(fdVar)) + + if err := os.Setenv(fdVar, value); err != nil { + fmt.Fprintf(os.Stderr, "can't set %s: %s\n", fdVar, err) + os.Exit(1) + } +} + +func getFileName(fd int) string { + fdVar := fdVarPrefix + fmt.Sprint(fd) + value := os.Getenv(fdVar) + sep := bytes.IndexByte([]byte(value), ':') + + if sep < 0 { + fmt.Fprintf(os.Stderr, "bad fd var %s\n", fdVar) + os.Exit(1) + } + name, err := hex.DecodeString(value[:sep]) + if err != nil { + fmt.Fprintf(os.Stderr, "error decoding %s\n", fdVar) + os.Exit(1) + } + return string(name) +} + +func resetFileName(fd int) { + fdVar := fdVarPrefix + fmt.Sprint(fd) + value := os.Getenv(fdVar) + sep := bytes.IndexByte([]byte(value), ':') + + if sep < 0 { + fmt.Fprintf(os.Stderr, "bad fd var %s\n", fdVar) + os.Exit(1) + } + if err := os.Setenv(fdVar, value[sep+1:]); err != nil { + fmt.Fprintf(os.Stderr, "can't reset %s\n", fdVar) + os.Exit(1) + } +} + +// DaemonStage tells in what stage in the process we are. See Stage(). +type DaemonStage int + +// Stages in the daemonizing process. +const ( + StageParent = DaemonStage(iota) // Original process + StageChild // MakeDaemon() called once: first child + StageDaemon // MakeDaemon() run twice: final daemon + + stageUnknown = DaemonStage(-1) +) + +var currStage = stageUnknown + +func Stage() DaemonStage { + if currStage == stageUnknown { + s, _, _ := getStage() + currStage = DaemonStage(s) + } + return currStage +} + +func (s DaemonStage) String() string { + switch s { + case StageParent: + return "parent" + case StageChild: + return "first child" + case StageDaemon: + return "daemon" + default: + return "unknown" + } +} + +func getStage() (stage int, advanceStage func() error, resetEnv func() error) { + var origValue string + stage = 0 + + daemonStage := os.Getenv(stageVar) + stageTag := strings.SplitN(daemonStage, ":", 2) + stageInfo := strings.SplitN(stageTag[0], "/", 3) + + if len(stageInfo) == 3 { + stageStr, tm, check := stageInfo[0], stageInfo[1], stageInfo[2] + + hash := sha1.New() + hash.Write([]byte(stageStr + "/" + tm + "/")) + + if check != hex.EncodeToString(hash.Sum([]byte{})) { + // This whole chunk is original data + origValue = daemonStage + } else { + stage, _ = strconv.Atoi(stageStr) + + if len(stageTag) == 2 { + origValue = stageTag[1] + } + } + } else { + origValue = daemonStage + } + + advanceStage = func() error { + base := fmt.Sprintf("%d/%09d/", stage+1, time.Now().Nanosecond()) + hash := sha1.New() + hash.Write([]byte(base)) + tag := base + hex.EncodeToString(hash.Sum([]byte{})) + + if err := os.Setenv(stageVar, tag+":"+origValue); err != nil { + return fmt.Errorf("can't set %s: %s", stageVar, err) + } + return nil + } + resetEnv = func() error { + return os.Setenv(stageVar, origValue) + } + + return stage, advanceStage, resetEnv +}