mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-06-13 23:36:45 +03:00
Chris Lu
56c51e7f50
CreateEntry is a FindEntry-then-write with no lock, so concurrent creates to the same path race: OExcl can admit two creators, and a conditional write has no atomic check-then-act. Add a per-path exclusive lock (util.LockTable, which evicts idle keys so it stays bounded) in the CreateEntry handler so the read and the write are atomic on this filer. Once callers route a key's writes to its owner filer, this local lock is the authoritative serialization point. AppendToEntry moves from the distributed lock to the same per-path lock.
324 lines
12 KiB
Go
324 lines
12 KiB
Go
package weed_server
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"net/http"
|
|
"os"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/credential"
|
|
"github.com/seaweedfs/seaweedfs/weed/stats"
|
|
"golang.org/x/sync/singleflight"
|
|
|
|
"google.golang.org/grpc"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/util/grace"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/operation"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/util"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/filer"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/filer/arangodb"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/filer/cassandra"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/filer/cassandra2"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/filer/elastic/v7"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/filer/etcd"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/filer/foundationdb"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/filer/hbase"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/filer/leveldb"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/filer/leveldb2"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/filer/leveldb3"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/filer/mongodb"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/filer/mysql"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/filer/mysql2"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/filer/postgres"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/filer/postgres2"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/filer/redis"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/filer/redis2"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/filer/redis3"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/filer/sqlite"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/filer/tarantool"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/filer/ydb"
|
|
"github.com/seaweedfs/seaweedfs/weed/glog"
|
|
"github.com/seaweedfs/seaweedfs/weed/notification"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/notification/aws_sqs"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/notification/gocdk_pub_sub"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/notification/google_pub_sub"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/notification/kafka"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/notification/log"
|
|
_ "github.com/seaweedfs/seaweedfs/weed/notification/webhook"
|
|
"github.com/seaweedfs/seaweedfs/weed/security"
|
|
)
|
|
|
|
type FilerOption struct {
|
|
Masters *pb.ServerDiscovery
|
|
FilerGroup string
|
|
Collection string
|
|
DefaultReplication string
|
|
DisableDirListing bool
|
|
MaxMB int
|
|
DirListingLimit int
|
|
DataCenter string
|
|
Rack string
|
|
DataNode string
|
|
DefaultLevelDbDir string
|
|
DisableHttp bool
|
|
Host pb.ServerAddress
|
|
recursiveDelete bool
|
|
Cipher bool
|
|
SaveToFilerLimit int64
|
|
ConcurrentUploadLimit int64
|
|
ConcurrentFileUploadLimit int64
|
|
ShowUIDirectoryDelete bool
|
|
DownloadMaxBytesPs int64
|
|
DiskType string
|
|
AllowedOrigins []string
|
|
ExposeDirectoryData bool
|
|
TusBasePath string
|
|
S3ConfigFile string // optional path to static S3 identity config file
|
|
CredentialManager *credential.CredentialManager
|
|
}
|
|
|
|
type FilerServer struct {
|
|
inFlightDataSize int64
|
|
inFlightUploads int64
|
|
listenersWaits int64
|
|
|
|
// notifying clients
|
|
listenersLock sync.Mutex
|
|
listenersCond *sync.Cond
|
|
|
|
inFlightDataLimitCond *sync.Cond
|
|
|
|
filer_pb.UnimplementedSeaweedFilerServer
|
|
option *FilerOption
|
|
filer *filer.Filer
|
|
filerGuard *security.Guard
|
|
volumeGuard *security.Guard
|
|
grpcDialOption grpc.DialOption
|
|
|
|
// metrics read from the master
|
|
metricsAddress string
|
|
metricsIntervalSec int
|
|
|
|
// track known metadata listeners
|
|
knownListenersLock sync.Mutex
|
|
knownListeners map[int32]int32
|
|
|
|
// deduplicates concurrent remote object caching operations
|
|
remoteCacheGroup singleflight.Group
|
|
|
|
recentCopyRequestsMu sync.Mutex
|
|
recentCopyRequests map[string]recentCopyRequest
|
|
|
|
// credential manager for IAM operations
|
|
CredentialManager *credential.CredentialManager
|
|
|
|
// mountPeerRegistry backs the MountRegister / MountList RPCs for peer
|
|
// chunk sharing (tier 1). Always populated.
|
|
mountPeerRegistry *filer.MountPeerRegistry
|
|
|
|
// entryLockTable serializes mutations to the same entry path on this filer.
|
|
// It is the local serialization point for read-modify-write operations
|
|
// (conditional create, append) once writers for a key are routed to this
|
|
// node, replacing the distributed lock for that purpose. Idle keys are
|
|
// evicted automatically, so the table stays bounded.
|
|
entryLockTable *util.LockTable[util.FullPath]
|
|
}
|
|
|
|
func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption) (fs *FilerServer, err error) {
|
|
|
|
v := util.GetViper()
|
|
signingKey := v.GetString("jwt.filer_signing.key")
|
|
v.SetDefault("jwt.filer_signing.expires_after_seconds", 10)
|
|
expiresAfterSec := v.GetInt("jwt.filer_signing.expires_after_seconds")
|
|
|
|
readSigningKey := v.GetString("jwt.filer_signing.read.key")
|
|
v.SetDefault("jwt.filer_signing.read.expires_after_seconds", 60)
|
|
readExpiresAfterSec := v.GetInt("jwt.filer_signing.read.expires_after_seconds")
|
|
|
|
volumeSigningKey := v.GetString("jwt.signing.key")
|
|
v.SetDefault("jwt.signing.expires_after_seconds", 10)
|
|
volumeExpiresAfterSec := v.GetInt("jwt.signing.expires_after_seconds")
|
|
|
|
volumeReadSigningKey := v.GetString("jwt.signing.read.key")
|
|
v.SetDefault("jwt.signing.read.expires_after_seconds", 60)
|
|
volumeReadExpiresAfterSec := v.GetInt("jwt.signing.read.expires_after_seconds")
|
|
|
|
v.SetDefault("cors.allowed_origins.values", "*")
|
|
|
|
allowedOrigins := v.GetString("cors.allowed_origins.values")
|
|
domains := strings.Split(allowedOrigins, ",")
|
|
option.AllowedOrigins = domains
|
|
|
|
v.SetDefault("filer.expose_directory_metadata.enabled", true)
|
|
returnDirMetadata := v.GetBool("filer.expose_directory_metadata.enabled")
|
|
option.ExposeDirectoryData = returnDirMetadata
|
|
|
|
fs = &FilerServer{
|
|
option: option,
|
|
grpcDialOption: security.LoadClientTLS(util.GetViper(), "grpc.filer"),
|
|
knownListeners: make(map[int32]int32),
|
|
inFlightDataLimitCond: sync.NewCond(new(sync.Mutex)),
|
|
recentCopyRequests: make(map[string]recentCopyRequest),
|
|
CredentialManager: option.CredentialManager,
|
|
entryLockTable: util.NewLockTable[util.FullPath](),
|
|
}
|
|
fs.mountPeerRegistry = filer.NewMountPeerRegistry()
|
|
go fs.runMountPeerRegistrySweeper()
|
|
fs.listenersCond = sync.NewCond(&fs.listenersLock)
|
|
|
|
option.Masters.RefreshBySrvIfAvailable()
|
|
if len(option.Masters.GetInstances()) == 0 {
|
|
glog.Fatal("master list is required!")
|
|
}
|
|
|
|
if !util.LoadConfiguration("filer", false) {
|
|
v.SetDefault("leveldb2.enabled", true)
|
|
v.SetDefault("leveldb2.dir", option.DefaultLevelDbDir)
|
|
_, err := os.Stat(option.DefaultLevelDbDir)
|
|
if os.IsNotExist(err) {
|
|
os.MkdirAll(option.DefaultLevelDbDir, 0755)
|
|
}
|
|
glog.V(0).Infof("default to create filer store dir in %s", option.DefaultLevelDbDir)
|
|
} else {
|
|
glog.Warningf("skipping default store dir in %s", option.DefaultLevelDbDir)
|
|
}
|
|
util.LoadConfiguration("notification", false)
|
|
|
|
v.SetDefault("filer.options.max_file_name_length", 255)
|
|
maxFilenameLength := v.GetUint32("filer.options.max_file_name_length")
|
|
glog.V(0).Infof("max_file_name_length %d", maxFilenameLength)
|
|
fs.filer = filer.NewFiler(*option.Masters, fs.grpcDialOption, option.Host, option.FilerGroup, option.Collection, option.DefaultReplication, option.DataCenter, maxFilenameLength, func() {
|
|
if atomic.LoadInt64(&fs.listenersWaits) > 0 {
|
|
fs.listenersCond.Broadcast()
|
|
}
|
|
})
|
|
fs.filer.Cipher = option.Cipher
|
|
// we do not support IP whitelist right now https://github.com/seaweedfs/seaweedfs/issues/7094
|
|
if v.GetString("guard.white_list") != "" {
|
|
glog.Warningf("filer: guard.white_list is configured but the IP whitelist feature is currently disabled. See https://github.com/seaweedfs/seaweedfs/issues/7094")
|
|
}
|
|
fs.filerGuard = security.NewGuard([]string{}, signingKey, expiresAfterSec, readSigningKey, readExpiresAfterSec)
|
|
fs.volumeGuard = security.NewGuard([]string{}, volumeSigningKey, volumeExpiresAfterSec, volumeReadSigningKey, volumeReadExpiresAfterSec)
|
|
|
|
fs.checkWithMaster()
|
|
|
|
go stats.LoopPushingMetric("filer", string(fs.option.Host), fs.metricsAddress, fs.metricsIntervalSec)
|
|
go fs.filer.MasterClient.KeepConnectedToMaster(context.Background())
|
|
|
|
fs.option.recursiveDelete = v.GetBool("filer.options.recursive_delete")
|
|
v.SetDefault("filer.options.buckets_folder", "/buckets")
|
|
fs.filer.DirBucketsPath = v.GetString("filer.options.buckets_folder")
|
|
// TODO deprecated, will be removed after 2020-12-31
|
|
// replaced by https://github.com/seaweedfs/seaweedfs/wiki/Path-Specific-Configuration
|
|
// fs.filer.FsyncBuckets = v.GetStringSlice("filer.options.buckets_fsync")
|
|
isFresh := fs.filer.LoadConfiguration(v)
|
|
|
|
notification.LoadConfiguration(v, "notification.")
|
|
|
|
handleStaticResources(defaultMux)
|
|
if !option.DisableHttp {
|
|
defaultMux.HandleFunc("/healthz", requestIDMiddleware(fs.filerHealthzHandler))
|
|
// TUS resumable upload protocol handler
|
|
if option.TusBasePath != "" {
|
|
// Normalize TusPath to always have a leading slash and no trailing slash
|
|
if !strings.HasPrefix(option.TusBasePath, "/") {
|
|
option.TusBasePath = "/" + option.TusBasePath
|
|
}
|
|
option.TusBasePath = strings.TrimRight(option.TusBasePath, "/")
|
|
|
|
// Disallow using "/" as TUS base to avoid hijacking all filer routes
|
|
if option.TusBasePath == "" {
|
|
glog.Warningf("Invalid TUS base path; TUS disabled (must not be root '/')")
|
|
} else {
|
|
handlePath := option.TusBasePath + "/"
|
|
defaultMux.HandleFunc(handlePath, fs.filerGuard.WhiteList(requestIDMiddleware(fs.tusHandler)))
|
|
// Start background cleanup of expired TUS sessions (every hour)
|
|
fs.StartTusSessionCleanup(1 * time.Hour)
|
|
}
|
|
}
|
|
defaultMux.HandleFunc("/", fs.filerGuard.WhiteList(requestIDMiddleware(fs.filerHandler)))
|
|
}
|
|
if defaultMux != readonlyMux {
|
|
handleStaticResources(readonlyMux)
|
|
readonlyMux.HandleFunc("/healthz", requestIDMiddleware(fs.filerHealthzHandler))
|
|
readonlyMux.HandleFunc("/", fs.filerGuard.WhiteList(requestIDMiddleware(fs.readonlyFilerHandler)))
|
|
}
|
|
|
|
existingNodes := fs.filer.ListExistingPeerUpdates(context.Background())
|
|
startFromTime := time.Now().Add(-filer.LogFlushInterval)
|
|
if isFresh {
|
|
glog.V(0).Infof("%s bootstrap from peers %+v", option.Host, existingNodes)
|
|
if err := fs.filer.MaybeBootstrapFromOnePeer(option.Host, existingNodes, startFromTime); err != nil {
|
|
glog.Fatalf("%s bootstrap from %+v: %v", option.Host, existingNodes, err)
|
|
}
|
|
}
|
|
v.SetDefault("filer.options.s3.empty_folder_cleanup_delay", "2m")
|
|
if d, err := time.ParseDuration(v.GetString("filer.options.s3.empty_folder_cleanup_delay")); err == nil {
|
|
fs.filer.EmptyFolderCleanupDelay = d
|
|
}
|
|
fs.filer.AggregateFromPeers(option.Host, existingNodes, startFromTime)
|
|
|
|
fs.filer.LoadFilerConf()
|
|
|
|
fs.filer.LoadRemoteStorageConfAndMapping()
|
|
|
|
grace.OnReload(fs.Reload)
|
|
|
|
fs.SetupDlmReplication()
|
|
fs.filer.Dlm.LockRing.SetTakeSnapshotCallback(fs.OnDlmChangeSnapshot)
|
|
|
|
if fs.CredentialManager != nil {
|
|
fs.CredentialManager.SetFilerAddressFunc(func() pb.ServerAddress {
|
|
return fs.option.Host
|
|
}, fs.grpcDialOption)
|
|
fs.CredentialManager.SetMasterClient(fs.filer.MasterClient, fs.grpcDialOption)
|
|
}
|
|
|
|
return fs, nil
|
|
}
|
|
|
|
func (fs *FilerServer) checkWithMaster() {
|
|
|
|
isConnected := false
|
|
for !isConnected {
|
|
fs.option.Masters.RefreshBySrvIfAvailable()
|
|
for _, master := range fs.option.Masters.GetInstances() {
|
|
readErr := operation.WithMasterServerClient(false, master, fs.grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
|
|
resp, err := masterClient.GetMasterConfiguration(context.Background(), &master_pb.GetMasterConfigurationRequest{})
|
|
if err != nil {
|
|
return fmt.Errorf("get master %s configuration: %v", master, err)
|
|
}
|
|
fs.metricsAddress, fs.metricsIntervalSec = resp.MetricsAddress, int(resp.MetricsIntervalSeconds)
|
|
return nil
|
|
})
|
|
if readErr == nil {
|
|
isConnected = true
|
|
} else {
|
|
time.Sleep(7 * time.Second)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Shutdown gracefully shuts down the filer server by waiting for in-flight uploads to complete.
|
|
// This prevents data corruption when the process receives SIGTERM during active uploads.
|
|
func (fs *FilerServer) Shutdown() {
|
|
glog.V(0).Infof("Shutting down filer")
|
|
fs.filer.Shutdown()
|
|
}
|
|
|
|
func (fs *FilerServer) Reload() {
|
|
glog.V(0).Infoln("Reload filer server...")
|
|
|
|
util.LoadConfiguration("security", false)
|
|
}
|