seaweedfs/weed/command/mount_std.go

264 lines
8.1 KiB
Go
Raw Normal View History

2022-02-14 21:48:48 +00:00
//go:build linux || darwin
// +build linux darwin
2022-02-11 04:32:13 +00:00
package command
import (
2022-02-11 06:43:55 +00:00
"context"
2022-02-11 04:32:13 +00:00
"fmt"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/mount"
2022-02-14 09:09:31 +00:00
"github.com/chrislusf/seaweedfs/weed/mount/meta_cache"
2022-02-11 04:46:53 +00:00
"github.com/chrislusf/seaweedfs/weed/mount/unmount"
2022-02-11 06:43:55 +00:00
"github.com/chrislusf/seaweedfs/weed/pb"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
2022-04-02 22:14:37 +00:00
"github.com/chrislusf/seaweedfs/weed/pb/mount_pb"
2022-02-11 06:43:55 +00:00
"github.com/chrislusf/seaweedfs/weed/security"
"github.com/chrislusf/seaweedfs/weed/storage/types"
"github.com/hanwen/go-fuse/v2/fuse"
2022-04-02 22:14:37 +00:00
"google.golang.org/grpc/reflection"
"net"
2022-02-11 04:32:13 +00:00
"net/http"
"os"
2022-02-11 06:43:55 +00:00
"os/user"
"runtime"
2022-02-11 04:32:13 +00:00
"strconv"
2022-02-11 06:43:55 +00:00
"strings"
2022-02-11 04:32:13 +00:00
"time"
"github.com/chrislusf/seaweedfs/weed/util"
"github.com/chrislusf/seaweedfs/weed/util/grace"
)
2022-02-27 10:57:27 +00:00
func runMount(cmd *Command, args []string) bool {
2022-02-11 04:32:13 +00:00
2022-02-27 10:57:27 +00:00
if *mountOptions.debug {
go http.ListenAndServe(fmt.Sprintf(":%d", *mountOptions.debugPort), nil)
2022-02-11 04:32:13 +00:00
}
grace.SetupProfiling(*mountCpuProfile, *mountMemProfile)
if *mountReadRetryTime < time.Second {
*mountReadRetryTime = time.Second
}
util.RetryWaitTime = *mountReadRetryTime
2022-02-27 10:57:27 +00:00
umask, umaskErr := strconv.ParseUint(*mountOptions.umaskString, 8, 64)
2022-02-11 04:32:13 +00:00
if umaskErr != nil {
2022-02-27 10:57:27 +00:00
fmt.Printf("can not parse umask %s", *mountOptions.umaskString)
2022-02-11 04:32:13 +00:00
return false
}
if len(args) > 0 {
return false
}
return RunMount(&mountOptions, os.FileMode(umask))
2022-02-11 04:32:13 +00:00
}
func RunMount(option *MountOptions, umask os.FileMode) bool {
2022-02-11 04:32:13 +00:00
2022-02-11 06:43:55 +00:00
// basic checks
2022-02-27 10:57:27 +00:00
chunkSizeLimitMB := *mountOptions.chunkSizeLimitMB
2022-02-11 06:43:55 +00:00
if chunkSizeLimitMB <= 0 {
fmt.Printf("Please specify a reasonable buffer size.")
return false
}
// try to connect to filer
filerAddresses := pb.ServerAddresses(*option.filer).ToAddresses()
util.LoadConfiguration("security", false)
grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.client")
var cipher bool
var err error
for i := 0; i < 10; i++ {
err = pb.WithOneOfGrpcFilerClients(false, filerAddresses, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
if err != nil {
return fmt.Errorf("get filer grpc address %v configuration: %v", filerAddresses, err)
}
cipher = resp.Cipher
return nil
})
if err != nil {
glog.V(0).Infof("failed to talk to filer %v: %v", filerAddresses, err)
glog.V(0).Infof("wait for %d seconds ...", i+1)
time.Sleep(time.Duration(i+1) * time.Second)
}
}
if err != nil {
glog.Errorf("failed to talk to filer %v: %v", filerAddresses, err)
return true
}
filerMountRootPath := *option.filerMountRootPath
// clean up mount point
dir := util.ResolvePath(*option.dir)
if dir == "" {
fmt.Printf("Please specify the mount directory via \"-dir\"")
return false
}
unmount.Unmount(dir)
2022-04-02 22:14:37 +00:00
// start on local unix socket
if *option.localSocket == "" {
mountDirHash := util.HashToInt32([]byte(dir))
if mountDirHash < 0 {
mountDirHash = -mountDirHash
}
*option.localSocket = fmt.Sprintf("/tmp/seaweefs-mount-%d.sock", mountDirHash)
2022-04-07 07:33:13 +00:00
}
if err := os.Remove(*option.localSocket); err != nil && !os.IsNotExist(err) {
glog.Fatalf("Failed to remove %s, error: %s", *option.localSocket, err.Error())
2022-04-02 22:14:37 +00:00
}
montSocketListener, err := net.Listen("unix", *option.localSocket)
if err != nil {
glog.Fatalf("Failed to listen on %s: %v", *option.localSocket, err)
}
2022-02-11 06:43:55 +00:00
// detect mount folder mode
if *option.dirAutoCreate {
os.MkdirAll(dir, os.FileMode(0777)&^umask)
}
fileInfo, err := os.Stat(dir)
// collect uid, gid
uid, gid := uint32(0), uint32(0)
mountMode := os.ModeDir | 0777
if err == nil {
mountMode = os.ModeDir | os.FileMode(0777)&^umask
uid, gid = util.GetFileUidGid(fileInfo)
fmt.Printf("mount point owner uid=%d gid=%d mode=%s\n", uid, gid, mountMode)
} else {
fmt.Printf("can not stat %s\n", dir)
return false
}
// detect uid, gid
if uid == 0 {
if u, err := user.Current(); err == nil {
if parsedId, pe := strconv.ParseUint(u.Uid, 10, 32); pe == nil {
uid = uint32(parsedId)
}
if parsedId, pe := strconv.ParseUint(u.Gid, 10, 32); pe == nil {
gid = uint32(parsedId)
}
fmt.Printf("current uid=%d gid=%d\n", uid, gid)
}
}
// mapping uid, gid
uidGidMapper, err := meta_cache.NewUidGidMapper(*option.uidMap, *option.gidMap)
if err != nil {
fmt.Printf("failed to parse %s %s: %v\n", *option.uidMap, *option.gidMap, err)
return false
}
// Ensure target mount point availability
if isValid := checkMountPointAvailable(dir); !isValid {
glog.Fatalf("Expected mount to still be active, target mount point: %s, please check!", dir)
return true
}
serverFriendlyName := strings.ReplaceAll(*option.filer, ",", "+")
2022-02-11 06:43:55 +00:00
// mount fuse
2022-02-12 05:35:09 +00:00
fuseMountOptions := &fuse.MountOptions{
AllowOther: *option.allowOthers,
Options: nil,
MaxBackground: 128,
MaxWrite: 1024 * 1024 * 2,
MaxReadAhead: 1024 * 1024 * 2,
IgnoreSecurityLabels: false,
RememberInodes: false,
FsName: serverFriendlyName + ":" + filerMountRootPath,
2022-02-12 05:35:09 +00:00
Name: "seaweedfs",
SingleThreaded: false,
DisableXAttrs: false,
2022-02-27 10:02:30 +00:00
Debug: *option.debug,
2022-02-12 05:35:09 +00:00
EnableLocks: false,
ExplicitDataCacheControl: false,
2022-02-27 09:13:32 +00:00
DirectMount: true,
DirectMountFlags: 0,
//SyncRead: false, // set to false to enable the FUSE_CAP_ASYNC_READ capability
//EnableAcl: true,
}
if *option.nonempty {
fuseMountOptions.Options = append(fuseMountOptions.Options, "nonempty")
}
if *option.readOnly {
if runtime.GOOS == "darwin" {
fuseMountOptions.Options = append(fuseMountOptions.Options, "rdonly")
} else {
fuseMountOptions.Options = append(fuseMountOptions.Options, "ro")
}
}
if runtime.GOOS == "darwin" {
// https://github-wiki-see.page/m/macfuse/macfuse/wiki/Mount-Options
ioSizeMB := 1
for ioSizeMB*2 <= *option.chunkSizeLimitMB && ioSizeMB*2 <= 32 {
ioSizeMB *= 2
}
fuseMountOptions.Options = append(fuseMountOptions.Options, "daemon_timeout=600")
fuseMountOptions.Options = append(fuseMountOptions.Options, "noapplexattr")
2022-02-27 10:02:30 +00:00
// fuseMountOptions.Options = append(fuseMountOptions.Options, "novncache") // need to test effectiveness
2022-02-27 09:13:32 +00:00
fuseMountOptions.Options = append(fuseMountOptions.Options, "slow_statfs")
fuseMountOptions.Options = append(fuseMountOptions.Options, "volname="+serverFriendlyName)
2022-02-27 09:13:32 +00:00
fuseMountOptions.Options = append(fuseMountOptions.Options, fmt.Sprintf("iosize=%d", ioSizeMB*1024*1024))
2022-02-12 05:35:09 +00:00
}
2022-02-11 04:46:53 +00:00
2022-02-11 06:43:55 +00:00
// find mount point
mountRoot := filerMountRootPath
if mountRoot != "/" && strings.HasSuffix(mountRoot, "/") {
mountRoot = mountRoot[0 : len(mountRoot)-1]
}
seaweedFileSystem := mount.NewSeaweedFileSystem(&mount.Option{
MountDirectory: dir,
FilerAddresses: filerAddresses,
GrpcDialOption: grpcDialOption,
FilerMountRootPath: mountRoot,
Collection: *option.collection,
Replication: *option.replication,
TtlSec: int32(*option.ttlSec),
DiskType: types.ToDiskType(*option.diskType),
ChunkSizeLimit: int64(chunkSizeLimitMB) * 1024 * 1024,
ConcurrentWriters: *option.concurrentWriters,
CacheDir: *option.cacheDir,
CacheSizeMB: *option.cacheSizeMB,
DataCenter: *option.dataCenter,
Quota: int64(*option.collectionQuota) * 1024 * 1024,
2022-02-11 06:43:55 +00:00
MountUid: uid,
MountGid: gid,
MountMode: mountMode,
MountCtime: fileInfo.ModTime(),
MountMtime: time.Now(),
Umask: umask,
2022-02-27 10:57:27 +00:00
VolumeServerAccess: *mountOptions.volumeServerAccess,
2022-02-11 06:43:55 +00:00
Cipher: cipher,
UidGidMapper: uidGidMapper,
2022-02-11 04:46:53 +00:00
})
2022-02-12 05:35:09 +00:00
server, err := fuse.NewServer(seaweedFileSystem, dir, fuseMountOptions)
2022-02-11 04:32:13 +00:00
if err != nil {
glog.Fatalf("Mount fail: %v", err)
}
2022-02-11 06:43:55 +00:00
grace.OnInterrupt(func() {
unmount.Unmount(dir)
})
2022-04-02 22:14:37 +00:00
grpcS := pb.NewGrpcServer()
mount_pb.RegisterSeaweedMountServer(grpcS, seaweedFileSystem)
reflection.Register(grpcS)
go grpcS.Serve(montSocketListener)
2022-02-14 09:09:31 +00:00
seaweedFileSystem.StartBackgroundTasks()
2022-02-11 06:43:55 +00:00
fmt.Printf("This is SeaweedFS version %s %s %s\n", util.Version(), runtime.GOOS, runtime.GOARCH)
2022-02-12 05:35:09 +00:00
server.Serve()
2022-02-11 04:32:13 +00:00
return true
}