From 9cae95d7496e805d7130bb918638e0733802cec1 Mon Sep 17 00:00:00 2001 From: os-pradipbabar Date: Sun, 12 Apr 2026 09:48:22 +0530 Subject: [PATCH] fix(filer): prevent data corruption during graceful shutdown (#9037) * 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 --- weed/command/filer.go | 91 +++++++++++++++++++++++++++++++++---- weed/server/filer_server.go | 10 ++-- 2 files changed, 88 insertions(+), 13 deletions(-) diff --git a/weed/command/filer.go b/weed/command/filer.go index 75d3ccfd4..1fa70dcde 100644 --- a/weed/command/filer.go +++ b/weed/command/filer.go @@ -11,6 +11,7 @@ import ( "runtime" "sort" "strings" + "sync" "time" "github.com/spf13/viper" @@ -31,6 +32,7 @@ import ( weed_server "github.com/seaweedfs/seaweedfs/weed/server" stats_collect "github.com/seaweedfs/seaweedfs/weed/stats" "github.com/seaweedfs/seaweedfs/weed/util" + "github.com/seaweedfs/seaweedfs/weed/util/grace" "github.com/seaweedfs/seaweedfs/weed/util/version" ) @@ -381,6 +383,15 @@ func (fo *FilerOptions) startFiler() { glog.Fatalf("Filer startup error: %v", nfs_err) } + // Ensure fs.Shutdown() runs exactly once, whether triggered by a signal hook + // or by the main goroutine after Serve() returns (e.g., MiniCluster tests). + var shutdownOnce sync.Once + shutdownFiler := func() { + shutdownOnce.Do(func() { + fs.Shutdown() + }) + } + if *fo.publicPort != 0 { publicListeningAddress := util.JoinHostPort(*fo.bindIp, *fo.publicPort) glog.V(0).Infoln("Start Seaweed filer server", version.Version(), "public at", publicListeningAddress) @@ -434,6 +445,24 @@ func (fo *FilerOptions) startFiler() { go grpcS.Serve(grpcL) pb.ServeGrpcOnLocalSocket(grpcS, grpcPort) + // Register graceful shutdown for gRPC server to wait for active RPCs + grace.OnInterrupt(func() { + glog.V(0).Infof("Gracefully stopping gRPC server") + stopped := make(chan struct{}) + go func() { + grpcS.GracefulStop() + close(stopped) + }() + select { + case <-stopped: + glog.V(0).Infof("gRPC server stopped gracefully") + case <-time.After(10 * time.Second): + glog.V(0).Infof("gRPC server graceful stop timed out, forcing stop") + grpcS.Stop() + } + }) + + var socketServer *http.Server if runtime.GOOS != "windows" { localSocket := *fo.localSocket if localSocket == "" { @@ -442,14 +471,12 @@ func (fo *FilerOptions) startFiler() { if err := os.Remove(localSocket); err != nil && !os.IsNotExist(err) { glog.Fatalf("Failed to remove %s, error: %s", localSocket, err.Error()) } - go func() { - // start on local unix socket - filerSocketListener, err := net.Listen("unix", localSocket) - if err != nil { - glog.Fatalf("Failed to listen on %s: %v", localSocket, err) - } - newHttpServer(defaultMux, nil).Serve(filerSocketListener) - }() + filerSocketListener, err := net.Listen("unix", localSocket) + if err != nil { + glog.Fatalf("Failed to listen on %s: %v", localSocket, err) + } + socketServer = newHttpServer(defaultMux, nil) + go socketServer.Serve(filerSocketListener) } if viper.GetString("https.filer.key") != "" { @@ -489,14 +516,34 @@ func (fo *FilerOptions) startFiler() { security.FixTlsConfig(util.GetViper(), tlsConfig) + var localTLSServer *http.Server if filerLocalListener != nil { + localTLSServer = newHttpServer(defaultMux, tlsConfig) go func() { - if err := newHttpServer(defaultMux, tlsConfig).ServeTLS(filerLocalListener, "", ""); err != nil { + if err := localTLSServer.ServeTLS(filerLocalListener, "", ""); err != nil { glog.Errorf("Filer Fail to serve: %v", err) } }() } httpS := newHttpServer(defaultMux, tlsConfig) + + // Register shutdown hooks: stop all HTTP servers, then close filer database + grace.OnInterrupt(func() { + glog.V(0).Infof("Gracefully stopping all HTTP servers") + shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + if socketServer != nil { + socketServer.Shutdown(shutdownCtx) + } + if localTLSServer != nil { + localTLSServer.Shutdown(shutdownCtx) + } + if err := httpS.Shutdown(shutdownCtx); err != nil { + glog.Warningf("HTTPS server shutdown: %v", err) + } + }) + grace.OnInterrupt(shutdownFiler) + if MiniClusterCtx != nil { ctx := MiniClusterCtx go func() { @@ -508,15 +555,37 @@ func (fo *FilerOptions) startFiler() { if err := httpS.ServeTLS(filerListener, "", ""); err != nil && err != http.ErrServerClosed { glog.Fatalf("Filer Fail to serve: %v", err) } + // Close database after servers have stopped to prevent data corruption + shutdownFiler() } else { + var localHTTPServer *http.Server if filerLocalListener != nil { + localHTTPServer = newHttpServer(defaultMux, nil) go func() { - if err := newHttpServer(defaultMux, nil).Serve(filerLocalListener); err != nil { + if err := localHTTPServer.Serve(filerLocalListener); err != nil { glog.Errorf("Filer Fail to serve: %v", err) } }() } httpS := newHttpServer(defaultMux, nil) + + // Register shutdown hooks: stop all HTTP servers, then close filer database + grace.OnInterrupt(func() { + glog.V(0).Infof("Gracefully stopping all HTTP servers") + shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + if socketServer != nil { + socketServer.Shutdown(shutdownCtx) + } + if localHTTPServer != nil { + localHTTPServer.Shutdown(shutdownCtx) + } + if err := httpS.Shutdown(shutdownCtx); err != nil { + glog.Warningf("HTTP server shutdown: %v", err) + } + }) + grace.OnInterrupt(shutdownFiler) + if MiniClusterCtx != nil { ctx := MiniClusterCtx go func() { @@ -528,5 +597,7 @@ func (fo *FilerOptions) startFiler() { if err := httpS.Serve(filerListener); err != nil && err != http.ErrServerClosed { glog.Fatalf("Filer Fail to serve: %v", err) } + // Close database after servers have stopped to prevent data corruption + shutdownFiler() } } diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go index a5cbef8d1..0365401be 100644 --- a/weed/server/filer_server.go +++ b/weed/server/filer_server.go @@ -258,9 +258,6 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption) fs.filer.LoadRemoteStorageConfAndMapping() grace.OnReload(fs.Reload) - grace.OnInterrupt(func() { - fs.filer.Shutdown() - }) fs.SetupDlmReplication() fs.filer.Dlm.LockRing.SetTakeSnapshotCallback(fs.OnDlmChangeSnapshot) @@ -298,6 +295,13 @@ func (fs *FilerServer) checkWithMaster() { } } +// 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...")