diff --git a/weed/command/filer.go b/weed/command/filer.go index e58a672bd..14e6f9051 100644 --- a/weed/command/filer.go +++ b/weed/command/filer.go @@ -505,21 +505,7 @@ func (fo *FilerOptions) startFiler() { if gracefulTimeout <= 0 { gracefulTimeout = 15 * time.Second } - stopGrpcServer := 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(gracefulTimeout): - glog.V(0).Infof("gRPC server graceful stop timed out after %s, forcing stop", gracefulTimeout) - grpcS.Stop() - } - } + stopGrpcServer := func() { gracefulStopGrpc(grpcS, gracefulTimeout) } var socketServer *http.Server if runtime.GOOS != "windows" { @@ -640,26 +626,9 @@ func (fo *FilerOptions) startFiler() { // The filer leaves the lock ring first: peers and S3 gateways route keys to it // until the ring changes, and would hit refused connections once gRPC stops. func newFilerShutdown(leaveLockRing, stopGrpc, shutdownFiler func(), httpServers ...*http.Server) func() { + drain := newGracefulShutdown(stopGrpc, shutdownFiler, httpServers...) return sync.OnceFunc(func() { leaveLockRing() - shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) - defer cancel() - var drained sync.WaitGroup - drained.Add(1) - go func() { - defer drained.Done() - stopGrpc() - }() - for _, server := range httpServers { - drained.Add(1) - go func() { - defer drained.Done() - if err := server.Shutdown(shutdownCtx); err != nil { - glog.Warningf("filer HTTP shutdown: %v", err) - } - }() - } - drained.Wait() - shutdownFiler() + drain() }) } diff --git a/weed/command/s3.go b/weed/command/s3.go index 672d4da7d..8df38908b 100644 --- a/weed/command/s3.go +++ b/weed/command/s3.go @@ -83,10 +83,9 @@ type S3Options struct { readerCacheSizeMB *int64 allowUntrustedRemoteEndpoints *bool - // shutdownCtx, when non-nil, tells startS3Server/startIcebergServer to - // gracefully shut down their HTTP/gRPC servers once the ctx is cancelled. - // Used by weed mini to orchestrate an ordered shutdown; nil for standalone - // weed s3. + // shutdownCtx, when non-nil, tells startS3Server to gracefully shut down + // its HTTP/gRPC servers once the ctx is cancelled, in addition to on + // interrupt. Used by weed mini to orchestrate an ordered shutdown. shutdownCtx context.Context } @@ -395,16 +394,17 @@ func (s3opt *S3Options) startS3Server() bool { if s3ApiServer_err != nil { glog.Fatalf("S3 API Server startup error: %v", s3ApiServer_err) } - defer s3ApiServer.Shutdown() + + var httpServers []*http.Server // Start Iceberg REST Catalog server if enabled if *s3opt.portIceberg > 0 { - go s3opt.startIcebergServer(s3ApiServer) + httpServers = append(httpServers, s3opt.startIcebergServer(s3ApiServer)) } // Start Lance Namespace server if enabled if s3opt.portLance != nil && *s3opt.portLance > 0 { - go s3opt.startLanceServer(s3ApiServer) + httpServers = append(httpServers, s3opt.startLanceServer(s3ApiServer)) } if runtime.GOOS != "windows" { @@ -415,13 +415,14 @@ func (s3opt *S3Options) startS3Server() bool { if err := os.Remove(localSocket); err != nil && !os.IsNotExist(err) { glog.Fatalf("Failed to remove %s, error: %s", localSocket, err.Error()) } + s3SocketListener, err := net.Listen("unix", localSocket) + if err != nil { + glog.Fatalf("Failed to listen on %s: %v", localSocket, err) + } + socketServer := newHttpServer(router, nil) + httpServers = append(httpServers, socketServer) go func() { - // start on local unix socket - s3SocketListener, err := net.Listen("unix", localSocket) - if err != nil { - glog.Fatalf("Failed to listen on %s: %v", localSocket, err) - } - if err := newHttpServer(router, nil).Serve(s3SocketListener); err != nil && err != http.ErrServerClosed { + if err := socketServer.Serve(s3SocketListener); err != nil && err != http.ErrServerClosed { glog.Fatalf("Failed to start S3 http server: %v", err) } }() @@ -456,6 +457,7 @@ func (s3opt *S3Options) startS3Server() bool { } go grpcS.Serve(grpcL) pb.ServeGrpcOnLocalSocket(grpcS, grpcPort) + stopGrpcServer := func() { gracefulStopGrpc(grpcS, 15*time.Second) } if *s3opt.tlsPrivateKey != "" { // Check for port conflict when both HTTP and HTTPS are enabled on the same port @@ -496,23 +498,22 @@ func (s3opt *S3Options) startS3Server() bool { if *s3opt.portHttps == 0 { glog.V(0).Infof("Start Seaweed S3 API Server %s at https port %d", version.Version(), *s3opt.port) if s3ApiLocalListener != nil { + localServer := newHttpServer(router, tlsConfig) + httpServers = append(httpServers, localServer) go func() { - if err = newHttpServer(router, tlsConfig).ServeTLS(s3ApiLocalListener, "", ""); err != nil { + if err := localServer.ServeTLS(s3ApiLocalListener, "", ""); err != nil && err != http.ErrServerClosed { glog.Fatalf("S3 API Server Fail to serve: %v", err) } }() } httpS := newHttpServer(router, tlsConfig) - if s3opt.shutdownCtx != nil { - go func() { - <-s3opt.shutdownCtx.Done() - httpS.Shutdown(context.Background()) - grpcS.Stop() - }() - } + httpServers = append(httpServers, httpS) + shutdown := s3opt.newShutdown(stopGrpcServer, s3ApiServer, httpServers) if err = httpS.ServeTLS(s3ApiListener, "", ""); err != nil && err != http.ErrServerClosed { glog.Fatalf("S3 API Server Fail to serve: %v", err) } + // Serve returns when listeners close, before active requests finish. + shutdown() } else { glog.V(0).Infof("Start Seaweed S3 API Server %s at https port %d", version.Version(), *s3opt.portHttps) s3ApiListenerHttps, s3ApiLocalListenerHttps, err := util.NewIpAndLocalListeners( @@ -521,14 +522,18 @@ func (s3opt *S3Options) startS3Server() bool { glog.Fatalf("S3 API HTTPS listener on %s:%d error: %v", *s3opt.bindIp, *s3opt.portHttps, err) } if s3ApiLocalListenerHttps != nil { + localHttpsServer := newHttpServer(router, tlsConfig) + httpServers = append(httpServers, localHttpsServer) go func() { - if err = newHttpServer(router, tlsConfig).ServeTLS(s3ApiLocalListenerHttps, "", ""); err != nil { + if err := localHttpsServer.ServeTLS(s3ApiLocalListenerHttps, "", ""); err != nil && err != http.ErrServerClosed { glog.Fatalf("S3 API Server Fail to serve: %v", err) } }() } + httpsServer := newHttpServer(router, tlsConfig) + httpServers = append(httpServers, httpsServer) go func() { - if err = newHttpServer(router, tlsConfig).ServeTLS(s3ApiListenerHttps, "", ""); err != nil { + if err := httpsServer.ServeTLS(s3ApiListenerHttps, "", ""); err != nil && err != http.ErrServerClosed { glog.Fatalf("S3 API Server Fail to serve: %v", err) } }() @@ -537,31 +542,42 @@ func (s3opt *S3Options) startS3Server() bool { if *s3opt.tlsPrivateKey == "" || *s3opt.portHttps > 0 { glog.V(0).Infof("Start Seaweed S3 API Server %s at http port %d", version.Version(), *s3opt.port) if s3ApiLocalListener != nil { + localServer := newHttpServer(router, nil) + httpServers = append(httpServers, localServer) go func() { - if err = newHttpServer(router, nil).Serve(s3ApiLocalListener); err != nil { + if err := localServer.Serve(s3ApiLocalListener); err != nil && err != http.ErrServerClosed { glog.Fatalf("S3 API Server Fail to serve: %v", err) } }() } httpS := newHttpServer(router, nil) - if s3opt.shutdownCtx != nil { - go func() { - <-s3opt.shutdownCtx.Done() - httpS.Shutdown(context.Background()) - grpcS.Stop() - }() - } + httpServers = append(httpServers, httpS) + shutdown := s3opt.newShutdown(stopGrpcServer, s3ApiServer, httpServers) if err = httpS.Serve(s3ApiListener); err != nil && err != http.ErrServerClosed { glog.Fatalf("S3 API Server Fail to serve: %v", err) } + // Serve returns when listeners close, before active requests finish. + shutdown() } return true } +func (s3opt *S3Options) newShutdown(stopGrpc func(), s3ApiServer *s3api.S3ApiServer, httpServers []*http.Server) func() { + shutdown := newGracefulShutdown(stopGrpc, s3ApiServer.Shutdown, httpServers...) + grace.OnInterrupt(shutdown) + if s3opt.shutdownCtx != nil { + go func() { + <-s3opt.shutdownCtx.Done() + shutdown() + }() + } + return shutdown +} + // startIcebergServer starts the Iceberg REST Catalog server on a separate port. -func (s3opt *S3Options) startIcebergServer(s3ApiServer *s3api.S3ApiServer) { +func (s3opt *S3Options) startIcebergServer(s3ApiServer *s3api.S3ApiServer) *http.Server { icebergRouter := mux.NewRouter().SkipClean(true) // warehouse/parent query values may legally contain ';', which Go's // url.ParseQuery would otherwise drop @@ -591,12 +607,6 @@ func (s3opt *S3Options) startIcebergServer(s3ApiServer *s3api.S3ApiServer) { glog.V(0).Infof("Start Iceberg REST Catalog Server at http://%s", listenAddress) httpS := newHttpServer(icebergRouter, nil) - if s3opt.shutdownCtx != nil { - go func() { - <-s3opt.shutdownCtx.Done() - httpS.Shutdown(context.Background()) - }() - } // Serve on localhost as well if we're bound to a different interface if icebergLocalListener != nil { go func() { @@ -605,15 +615,18 @@ func (s3opt *S3Options) startIcebergServer(s3ApiServer *s3api.S3ApiServer) { } }() } - if err = httpS.Serve(icebergListener); err != nil && err != http.ErrServerClosed { - glog.Fatalf("Iceberg REST Catalog Server Fail to serve: %v", err) - } + go func() { + if err := httpS.Serve(icebergListener); err != nil && err != http.ErrServerClosed { + glog.Fatalf("Iceberg REST Catalog Server Fail to serve: %v", err) + } + }() + return httpS } // startLanceServer starts the Lance Namespace server on a separate port. It // shares the Iceberg catalog's credential role: one deployment vends table // credentials one way, whichever catalog the client speaks to. -func (s3opt *S3Options) startLanceServer(s3ApiServer *s3api.S3ApiServer) { +func (s3opt *S3Options) startLanceServer(s3ApiServer *s3api.S3ApiServer) *http.Server { lanceRouter := mux.NewRouter().SkipClean(true) lanceRouter.Use(util_http.EscapeSemicolonsInQuery) @@ -636,12 +649,6 @@ func (s3opt *S3Options) startLanceServer(s3ApiServer *s3api.S3ApiServer) { glog.V(0).Infof("Start Lance Namespace Server at http://%s", listenAddress) httpS := newHttpServer(lanceRouter, nil) - if s3opt.shutdownCtx != nil { - go func() { - <-s3opt.shutdownCtx.Done() - httpS.Shutdown(context.Background()) - }() - } if lanceLocalListener != nil { go func() { if err := httpS.Serve(lanceLocalListener); err != nil && err != http.ErrServerClosed { @@ -649,9 +656,12 @@ func (s3opt *S3Options) startLanceServer(s3ApiServer *s3api.S3ApiServer) { } }() } - if err = httpS.Serve(lanceListener); err != nil && err != http.ErrServerClosed { - glog.Fatalf("Lance Namespace Server Fail to serve: %v", err) - } + go func() { + if err := httpS.Serve(lanceListener); err != nil && err != http.ErrServerClosed { + glog.Fatalf("Lance Namespace Server Fail to serve: %v", err) + } + }() + return httpS } // deriveLanceStorageEndpoint picks the endpoint the Lance namespace puts in diff --git a/weed/command/s3_shutdown_test.go b/weed/command/s3_shutdown_test.go new file mode 100644 index 000000000..4b3126fd1 --- /dev/null +++ b/weed/command/s3_shutdown_test.go @@ -0,0 +1,81 @@ +package command + +import ( + "context" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/s3api" +) + +func TestS3ShutdownDrainsActiveRequestBeforeServeReturns(t *testing.T) { + entered := make(chan struct{}) + release := make(chan struct{}) + finish := sync.OnceFunc(func() { close(release) }) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + close(entered) + <-release + w.WriteHeader(http.StatusNoContent) + })) + t.Cleanup(server.Close) + t.Cleanup(finish) + httpClosing := make(chan struct{}) + server.Config.RegisterOnShutdown(func() { close(httpClosing) }) + + requestDone := make(chan struct{}) + go func() { + defer close(requestDone) + response, err := server.Client().Get(server.URL) + if err != nil { + t.Error(err) + return + } + response.Body.Close() + if response.StatusCode != http.StatusNoContent { + t.Errorf("active request returned %s", response.Status) + } + }() + select { + case <-entered: + case <-requestDone: + t.Fatal("request ended before reaching the handler") + case <-time.After(5 * time.Second): + t.Fatal("request did not reach the handler") + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + grpcStopped := make(chan struct{}) + s3opt := &S3Options{shutdownCtx: ctx} + shutdown := s3opt.newShutdown(func() { close(grpcStopped) }, &s3api.S3ApiServer{}, []*http.Server{server.Config}) + + cancel() + select { + case <-httpClosing: + case <-time.After(5 * time.Second): + t.Fatal("cancelling shutdownCtx did not start the HTTP drain") + } + <-grpcStopped + + // Model the main Serve path joining shutdown once its listener closes. + serveReturned := make(chan struct{}) + go func() { shutdown(); close(serveReturned) }() + select { + case <-serveReturned: + t.Fatal("Serve exit path returned while a request was still active") + case <-requestDone: + t.Fatal("active request ended before it was released") + case <-time.After(100 * time.Millisecond): + } + + finish() + <-requestDone + select { + case <-serveReturned: + case <-time.After(5 * time.Second): + t.Fatal("Serve exit path did not return after the request drained") + } +} diff --git a/weed/command/shutdown.go b/weed/command/shutdown.go new file mode 100644 index 000000000..18a17616a --- /dev/null +++ b/weed/command/shutdown.go @@ -0,0 +1,53 @@ +package command + +import ( + "context" + "net/http" + "sync" + "time" + + "google.golang.org/grpc" + + "github.com/seaweedfs/seaweedfs/weed/glog" +) + +func gracefulStopGrpc(grpcS *grpc.Server, timeout time.Duration) { + 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(timeout): + glog.V(0).Infof("gRPC server graceful stop timed out after %s, forcing stop", timeout) + grpcS.Stop() + } +} + +// newGracefulShutdown joins shutdown callers while gRPC and HTTP drain concurrently. +func newGracefulShutdown(stopGrpc, closeServer func(), httpServers ...*http.Server) func() { + return sync.OnceFunc(func() { + shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + var drained sync.WaitGroup + drained.Add(1) + go func() { + defer drained.Done() + stopGrpc() + }() + for _, server := range httpServers { + drained.Add(1) + go func() { + defer drained.Done() + if err := server.Shutdown(shutdownCtx); err != nil { + glog.Warningf("HTTP shutdown: %v", err) + } + }() + } + drained.Wait() + closeServer() + }) +} diff --git a/weed/command/shutdown_test.go b/weed/command/shutdown_test.go new file mode 100644 index 000000000..9dd142a77 --- /dev/null +++ b/weed/command/shutdown_test.go @@ -0,0 +1,96 @@ +package command + +import ( + "context" + "net" + "sync" + "testing" + "time" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + + "github.com/seaweedfs/seaweedfs/weed/pb/iam_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/s3_pb" +) + +type blockingIamCache struct { + s3_pb.UnimplementedSeaweedS3IamCacheServer + entered chan struct{} + release chan struct{} +} + +func (b *blockingIamCache) PutIdentity(ctx context.Context, _ *iam_pb.PutIdentityRequest) (*iam_pb.PutIdentityResponse, error) { + close(b.entered) + // Stop only cancels handler contexts; it cannot interrupt a handler that ignores them. + select { + case <-b.release: + return &iam_pb.PutIdentityResponse{}, nil + case <-ctx.Done(): + return nil, ctx.Err() + } +} + +func startBlockingGrpc(t *testing.T) (grpcS *grpc.Server, release func(), rpcErr chan error) { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + handler := &blockingIamCache{entered: make(chan struct{}), release: make(chan struct{})} + grpcS = grpc.NewServer() + s3_pb.RegisterSeaweedS3IamCacheServer(grpcS, handler) + go grpcS.Serve(listener) + t.Cleanup(grpcS.Stop) + release = sync.OnceFunc(func() { close(handler.release) }) + t.Cleanup(release) + + conn, err := grpc.NewClient(listener.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { conn.Close() }) + rpcErr = make(chan error, 1) + go func() { + _, err := s3_pb.NewSeaweedS3IamCacheClient(conn).PutIdentity(context.Background(), &iam_pb.PutIdentityRequest{}) + rpcErr <- err + }() + <-handler.entered + return grpcS, release, rpcErr +} + +func TestGracefulStopGrpcWaitsForInFlightRPC(t *testing.T) { + grpcS, release, rpcErr := startBlockingGrpc(t) + + stopped := make(chan struct{}) + go func() { gracefulStopGrpc(grpcS, 5*time.Second); close(stopped) }() + select { + case <-stopped: + t.Fatal("gRPC stopped while an RPC was in flight") + case <-time.After(100 * time.Millisecond): + } + release() + if err := <-rpcErr; err != nil { + t.Errorf("in-flight RPC failed: %v", err) + } + select { + case <-stopped: + case <-time.After(5 * time.Second): + t.Fatal("gRPC graceful stop did not return after the RPC completed") + } +} + +func TestGracefulStopGrpcForcesStopAfterTimeout(t *testing.T) { + grpcS, _, rpcErr := startBlockingGrpc(t) + + stopped := make(chan struct{}) + go func() { gracefulStopGrpc(grpcS, 50*time.Millisecond); close(stopped) }() + select { + case <-stopped: + case <-time.After(5 * time.Second): + t.Fatal("gRPC graceful stop did not honour its timeout") + } + if err := <-rpcErr; err == nil { + t.Error("in-flight RPC succeeded although the server was force-stopped") + } +}