mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-09-04 07:07:18 +00:00
Chris Lu
bd3b75b534
The inode table only shrinks when the kernel returns a reference, so a walk over a large tree grows it with every child it touches and nothing gives any of it back until the kernel reclaims dentries. On a mount with tens of millions of files that is most of the process's heap. -maxInodeEntries stops the speculative half of that. Past the cap a readdirplus still reports every name, but leaves the EntryOut zeroed: Linux reads a zero nodeid as 'no attributes for this entry', takes no reference, and looks up the ones the client actually needs. A LOOKUP is never refused, so the table can still grow past the cap by what a client really asks for, and nothing here evicts. Default 0 keeps today's behavior.
287 lines
10 KiB
Go
287 lines
10 KiB
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
|
|
|
|
// eagerFilerCreate persists a created file's entry at create time rather
|
|
// than at flush. A platform sets it when it cannot run the flush before
|
|
// the application's close returns, so nothing that reads through the
|
|
// filer can race an unflushed close.
|
|
eagerFilerCreate bool
|
|
}
|
|
|
|
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,
|
|
MaxInodeEntries: *option.maxInodeEntries,
|
|
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,
|
|
EagerFilerCreate: p.eagerFilerCreate,
|
|
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
|
|
}
|