seaweedfs/weed/command/mount_common.go

//go:build linux || darwin || freebsd || windows

package command

import (
	"context"
	"fmt"
	"net"
	"net/http"
	"os"
	"path"
	"runtime"
	"strconv"
	"strings"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/reflection"

	"github.com/seaweedfs/seaweedfs/weed/glog"
	"github.com/seaweedfs/seaweedfs/weed/mount"
	"github.com/seaweedfs/seaweedfs/weed/mount/meta_cache"
	"github.com/seaweedfs/seaweedfs/weed/pb"
	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
	"github.com/seaweedfs/seaweedfs/weed/pb/mount_pb"
	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
	"github.com/seaweedfs/seaweedfs/weed/security"
	"github.com/seaweedfs/seaweedfs/weed/storage/types"
	"github.com/seaweedfs/seaweedfs/weed/util"
	"github.com/seaweedfs/seaweedfs/weed/util/grace"
)

func runMount(cmd *Command, args []string) bool {

	if *mountOptions.debug {
		go http.ListenAndServe(fmt.Sprintf(":%d", *mountOptions.debugPort), nil)
	}

	*mountCpuProfile = util.ResolvePath(*mountCpuProfile)
	*mountMemProfile = util.ResolvePath(*mountMemProfile)
	grace.SetupProfiling(*mountCpuProfile, *mountMemProfile)
	if *mountReadRetryTime < time.Second {
		*mountReadRetryTime = time.Second
	}
	util.RetryWaitTime = *mountReadRetryTime

	// 32 bits, not 64: os.FileMode is uint32, so a wider parse would let a
	// nonsense umask truncate silently instead of being rejected here.
	umask, umaskErr := strconv.ParseUint(*mountOptions.umaskString, 8, 32)
	if umaskErr != nil {
		fmt.Printf("can not parse umask %s", *mountOptions.umaskString)
		return false
	}

	if len(args) > 0 {
		return false
	}

	return RunMount(&mountOptions, os.FileMode(umask))
}

func ensureBucketAllowEmptyFolders(ctx context.Context, filerClient filer_pb.FilerClient, mountRoot, bucketRootPath string) error {
	bucketPath, isBucketRootMount := bucketPathForMountRoot(mountRoot, bucketRootPath)
	if !isBucketRootMount {
		return nil
	}

	entry, _, _, err := filer_pb.GetEntry(ctx, filerClient, util.FullPath(bucketPath))
	if err != nil {
		return err
	}
	if entry == nil {
		return fmt.Errorf("bucket %s not found", bucketPath)
	}

	if entry.Extended == nil {
		entry.Extended = make(map[string][]byte)
	}
	if strings.EqualFold(strings.TrimSpace(string(entry.Extended[s3_constants.ExtAllowEmptyFolders])), "true") {
		return nil
	}

	entry.Extended[s3_constants.ExtAllowEmptyFolders] = []byte("true")

	bucketFullPath := util.FullPath(bucketPath)
	parent, _ := bucketFullPath.DirAndName()
	if err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
		return filer_pb.UpdateEntry(ctx, client, &filer_pb.UpdateEntryRequest{
			Directory: parent,
			Entry:     entry,
		})
	}); err != nil {
		return err
	}

	glog.V(3).Infof("RunMount: set bucket %s %s=true", bucketPath, s3_constants.ExtAllowEmptyFolders)
	return nil
}

func bucketPathForMountRoot(mountRoot, bucketRootPath string) (string, bool) {
	cleanPath := path.Clean("/" + strings.TrimPrefix(mountRoot, "/"))
	cleanBucketRoot := path.Clean("/" + strings.TrimPrefix(bucketRootPath, "/"))
	if cleanBucketRoot == "/" {
		return "", false
	}
	prefix := cleanBucketRoot + "/"
	if !strings.HasPrefix(cleanPath, prefix) {
		return "", false
	}
	rest := strings.TrimPrefix(cleanPath, prefix)

	bucketParts := strings.Split(rest, "/")
	if len(bucketParts) != 1 || bucketParts[0] == "" {
		return "", false
	}
	return cleanBucketRoot + "/" + bucketParts[0], true
}

func peerStringOrEmpty(p *string) string {
	if p == nil {
		return ""
	}
	return *p
}

// connectToFiler retries the filer handshake, returning the cluster's cipher
// setting and bucket root.
func connectToFiler(option *MountOptions) (filerAddresses []pb.ServerAddress, grpcDialOption grpc.DialOption, cipher bool, bucketRootPath string, ok bool) {
	// try to connect to filer
	filerAddresses = pb.ServerAddresses(*option.filer).ToAddresses()
	util.LoadSecurityConfiguration()
	grpcDialOption = security.LoadClientTLS(util.GetViper(), "grpc.client")
	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: %w", filerAddresses, err)
			}
			cipher = resp.Cipher
			bucketRootPath = resp.DirBuckets
			return nil
		})
		if err == nil {
			break
		}
		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 nil, nil, false, "", false
	}
	if bucketRootPath == "" {
		bucketRootPath = "/buckets"
	}
	return filerAddresses, grpcDialOption, cipher, bucketRootPath, true
}

// fileSystemParams are the pieces of a mount that each platform works out for
// itself: where it is attached, and whose identity the entries carry.
type fileSystemParams struct {
	dir              string
	mountRoot        string
	filerAddresses   []pb.ServerAddress
	grpcDialOption   grpc.DialOption
	cipher           bool
	uidGidMapper     *meta_cache.UidGidMapper
	uid              uint32
	gid              uint32
	mountMode        os.FileMode
	mountCtime       time.Time
	umask            os.FileMode
	chunkSizeLimitMB int
	cacheDirForRead  string
	cacheDirForWrite string
}

func buildSeaweedFileSystem(option *MountOptions, p fileSystemParams) *mount.WFS {
	return mount.NewSeaweedFileSystem(&mount.Option{
		MountDirectory:              p.dir,
		FilerAddresses:              p.filerAddresses,
		GrpcDialOption:              p.grpcDialOption,
		FilerSigningKey:             security.SigningKey(util.GetViper().GetString("jwt.filer_signing.key")),
		FilerSigningExpiresAfterSec: util.GetViper().GetInt("jwt.filer_signing.expires_after_seconds"),
		FilerMountRootPath:          p.mountRoot,
		Collection:                  *option.collection,
		Replication:                 *option.replication,
		TtlSec:                      int32(*option.ttlSec),
		DiskType:                    types.ToDiskType(*option.diskType),
		ChunkSizeLimit:              int64(p.chunkSizeLimitMB) * 1024 * 1024,
		ConcurrentWriters:           *option.concurrentWriters,
		ConcurrentReaders:           *option.concurrentReaders,
		CacheDirForRead:             p.cacheDirForRead,
		CacheSizeMBForRead:          *option.cacheSizeMBForRead,
		CacheDirForWrite:            p.cacheDirForWrite,
		WriteBufferSizeMB:           *option.writeBufferSizeMB,
		CacheMetaTTlSec:             *option.cacheMetaTtlSec,
		CacheDirMaxEntries:          *option.cacheDirMaxEntries,
		DataCenter:                  *option.dataCenter,
		Quota:                       int64(*option.collectionQuota) * 1024 * 1024,
		LogicalDiskUsage:            *option.logicalDiskUsage,
		MountUid:                    p.uid,
		MountGid:                    p.gid,
		MountMode:                   p.mountMode,
		MountCtime:                  p.mountCtime,
		MountMtime:                  time.Now(),
		Umask:                       p.umask,
		VolumeServerAccess:          *mountOptions.volumeServerAccess,
		Cipher:                      p.cipher,
		UidGidMapper:                p.uidGidMapper,
		IncludeSystemEntries:        *option.includeSystemEntries,
		DefaultPermissions:          *option.defaultPermissions,
		DisableXAttr:                *option.disableXAttr,
		IsMacOs:                     runtime.GOOS == "darwin",
		MetadataFlushSeconds:        *option.metadataFlushSeconds,
		// RDMA acceleration options
		RdmaEnabled:           *option.rdmaEnabled,
		RdmaSidecarAddr:       *option.rdmaSidecarAddr,
		RdmaFallback:          *option.rdmaFallback,
		RdmaReadOnly:          *option.rdmaReadOnly,
		RdmaMaxConcurrent:     *option.rdmaMaxConcurrent,
		RdmaTimeoutMs:         *option.rdmaTimeoutMs,
		DirIdleEvictSec:       *option.dirIdleEvictSec,
		EnableDistributedLock: option.distributedLock != nil && *option.distributedLock,
		WritebackCache:        option.writebackCache != nil && *option.writebackCache,
		PosixDirNlink:         option.posixDirNlink != nil && *option.posixDirNlink,
		// Peer chunk sharing
		PeerEnabled:    option.peerEnabled != nil && *option.peerEnabled,
		PeerListen:     peerStringOrEmpty(option.peerListen),
		PeerAdvertise:  peerStringOrEmpty(option.peerAdvertise),
		PeerDataCenter: peerStringOrEmpty(option.peerDataCenter),
		PeerRack:       peerStringOrEmpty(option.peerRack),
	})
}

// createMountRoot makes the filer-side directory the mount is rooted at.
func createMountRoot(wfs *mount.WFS, mountRoot, bucketRootPath string, filerAddresses []pb.ServerAddress) bool {
	mountRootPath := util.FullPath(mountRoot)
	mountRootParent, mountDir := mountRootPath.DirAndName()
	if err := filer_pb.Mkdir(context.Background(), wfs, mountRootParent, mountDir, nil); err != nil {
		fmt.Printf("failed to create dir %s on filer %s: %v\n", mountRoot, filerAddresses, err)
		return false
	}
	if err := ensureBucketAllowEmptyFolders(context.Background(), wfs, mountRoot, bucketRootPath); err != nil {
		fmt.Printf("failed to set bucket auto-remove-empty-folders policy for %s: %v\n", mountRoot, err)
		return false
	}
	return true
}

// serveMountGrpc exposes the local control socket used by "weed mount.stats".
func serveMountGrpc(wfs *mount.WFS, listener net.Listener) {
	grpcS := pb.NewGrpcServer()
	mount_pb.RegisterSeaweedMountServer(grpcS, wfs)
	reflection.Register(grpcS)
	go grpcS.Serve(listener)
}

// resolveMountRoot trims the trailing slash the filer path must not carry.
func resolveMountRoot(filerMountRootPath string) string {
	mountRoot := filerMountRootPath
	if mountRoot != "/" && strings.HasSuffix(mountRoot, "/") {
		mountRoot = mountRoot[0 : len(mountRoot)-1]
	}
	return mountRoot
}

// resolveCacheDirs falls back to the read cache when no write cache is set.
func resolveCacheDirs(option *MountOptions) (string, string) {
	cacheDirForRead := util.ResolvePath(*option.cacheDirForRead)
	cacheDirForWrite := util.ResolvePath(*option.cacheDirForWrite)
	if cacheDirForWrite == "" {
		cacheDirForWrite = cacheDirForRead
	}
	return cacheDirForRead, cacheDirForWrite
}