mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-08-01 12:56:33 +00:00
9cae95d749
* fix: wait for in-flight uploads to complete before filer shutdown Prevents data corruption when SIGTERM is received during active uploads. The filer now waits for all in-flight operations to complete before calling the underlying shutdown logic. This affects all deployment types (Kubernetes, Docker, systemd) and fixes corruption issues during rolling updates, certificate rotation, and manual restarts. Changes: - Add FilerServer.Shutdown() method with upload wait logic - Update grace.OnInterrupt hook to use new shutdown method Fixes data corruption reported by production users during pod restarts. * fix: implement graceful shutdown for gRPC and HTTP servers, ensuring in-flight uploads complete * fix: address review comments on graceful shutdown - Add 10s timeout to gRPC GracefulStop to prevent indefinite blocking from long-lived streams (falls back to Stop on timeout) - Reduce HTTP/HTTPS shutdown timeout from 25s to 15s to fit within Kubernetes default 30s termination grace period - Move fs.Shutdown() (database close) after Serve() returns instead of a separate hook to eliminate race where main goroutine exits before the shutdown hook runs * fix: shut down all HTTP servers before filer database close Address remaining review comments: - Shut down auxiliary HTTP servers (Unix socket, local listener) during graceful shutdown so they can't serve write traffic after the main server stops - Register fs.Shutdown() as a grace.OnInterrupt hook to guarantee it completes before os.Exit(0), fixing the race between the grace goroutine and the main goroutine - Use sync.Once to ensure fs.Shutdown() runs exactly once regardless of whether shutdown is signal-driven or context-driven (MiniCluster) --------- Co-authored-by: Chris Lu <chris.lu@gmail.com>
310 lines
12 KiB
Go
310 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
|
|
}
|
|
|
|
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,
|
|
}
|
|
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)
|
|
}
|