From 46ce2c45a2829c7bb1573a2090462f9824ffbb22 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 24 Aug 2026 14:55:53 -0700 Subject: [PATCH] mini: reserve the admin gRPC port instead of binding it late (#10928) * mini: reserve the admin gRPC port instead of binding it late Port selection probes every port with a throwaway listener and closes it. Master, filer, volume and S3 bind a moment later, but the admin waits for all of them first and only then binds its worker gRPC port, roughly two seconds in. That port defaults to the admin http port + 10000, which lands inside the Linux ephemeral range, so one of the cluster's own outgoing gRPC dials can take it during the gap and the admin dies on bind, taking the worker with it. Keep the listener from the availability check and hand it to the admin. * mini: clear the admin gRPC reservation before retaking it A rerun inside one process would otherwise inherit the closed listener of the previous run whenever the reservation fails, and the admin would accept it and only find out inside Serve. * mini: snapshot the admin options for the startup goroutine The cleanup path read the package-level options long after the goroutine started, so a later in-process run could have its reserved listener closed by the previous run. --- weed/admin/dash/admin_server.go | 5 +++-- weed/admin/dash/worker_grpc_server.go | 15 +++++++++------ weed/command/admin.go | 7 ++++++- weed/command/mini.go | 24 +++++++++++++++++++++++- 4 files changed, 41 insertions(+), 10 deletions(-) diff --git a/weed/admin/dash/admin_server.go b/weed/admin/dash/admin_server.go index f22490392..b5770db93 100644 --- a/weed/admin/dash/admin_server.go +++ b/weed/admin/dash/admin_server.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "io" + "net" "net/http" "sort" "strings" @@ -1589,13 +1590,13 @@ func (as *AdminServer) GetConfigInfo(w http.ResponseWriter, r *http.Request) { } // StartWorkerGrpcServer starts the worker gRPC server -func (s *AdminServer) StartWorkerGrpcServer(grpcPort int) error { +func (s *AdminServer) StartWorkerGrpcServer(grpcPort int, listener net.Listener) error { if s.workerGrpcServer != nil { return fmt.Errorf("worker gRPC server is already running") } s.workerGrpcServer = NewWorkerGrpcServer(s) - return s.workerGrpcServer.StartWithTLS(grpcPort) + return s.workerGrpcServer.StartWithTLS(grpcPort, listener) } // StopWorkerGrpcServer stops the worker gRPC server diff --git a/weed/admin/dash/worker_grpc_server.go b/weed/admin/dash/worker_grpc_server.go index 9fbce29c5..c6d9f4dba 100644 --- a/weed/admin/dash/worker_grpc_server.go +++ b/weed/admin/dash/worker_grpc_server.go @@ -83,16 +83,19 @@ func NewWorkerGrpcServer(adminServer *AdminServer) *WorkerGrpcServer { } } -// StartWithTLS starts the gRPC server on the specified port with optional TLS -func (s *WorkerGrpcServer) StartWithTLS(port int) error { +// StartWithTLS starts the gRPC server on the specified port with optional TLS. +// A caller that already holds the port passes its listener instead. +func (s *WorkerGrpcServer) StartWithTLS(port int, listener net.Listener) error { if s.running { return fmt.Errorf("worker gRPC server is already running") } - // Create listener - listener, err := net.Listen("tcp", fmt.Sprintf(":%d", port)) - if err != nil { - return fmt.Errorf("failed to listen on port %d: %v", port, err) + if listener == nil { + var err error + listener, err = net.Listen("tcp", fmt.Sprintf(":%d", port)) + if err != nil { + return fmt.Errorf("failed to listen on port %d: %v", port, err) + } } // Create gRPC server with optional TLS diff --git a/weed/command/admin.go b/weed/command/admin.go index 41bf74d91..5c462d466 100644 --- a/weed/command/admin.go +++ b/weed/command/admin.go @@ -59,6 +59,11 @@ type AdminOptions struct { debugPort *int cpuProfile *string memProfile *string + + // workerGrpcListener, when set, is a listener already bound to grpcPort by + // the caller. `weed mini` reserves the port this way because the admin + // binds it only after every other service is up. + workerGrpcListener net.Listener } func init() { @@ -406,7 +411,7 @@ func startAdminServer(ctx context.Context, options AdminOptions, enableUI bool, } // Start worker gRPC server for worker connections - err = adminServer.StartWorkerGrpcServer(*options.grpcPort) + err = adminServer.StartWorkerGrpcServer(*options.grpcPort, options.workerGrpcListener) if err != nil { return fmt.Errorf("failed to start worker gRPC server: %w", err) } diff --git a/weed/command/mini.go b/weed/command/mini.go index 0e9d8f45f..ed1725cde 100644 --- a/weed/command/mini.go +++ b/weed/command/mini.go @@ -942,6 +942,20 @@ func ensureAllPortsAvailableOnIP(bindIp string) error { // All gRPC port handling (calculation, validation, and assignment) is performed exclusively in initializeGrpcPortsOnIP initializeGrpcPortsOnIP(bindIp) + // Every other service binds within a moment of this check, but the admin + // waits for all of them first. The admin gRPC port sits inside the Linux + // ephemeral range, so during that gap one of the cluster's own outgoing + // connections can take it and the admin then dies on bind. Hold a listener + // from here and hand it to the admin instead of re-binding later. Clear + // first: an in-process rerun would otherwise inherit the closed listener + // of the previous run and only find out inside Serve. + miniAdminOptions.workerGrpcListener = nil + if listener, err := net.Listen("tcp", fmt.Sprintf(":%d", *miniAdminOptions.grpcPort)); err != nil { + glog.Warningf("Could not reserve Admin gRPC port %d: %v", *miniAdminOptions.grpcPort, err) + } else { + miniAdminOptions.workerGrpcListener = listener + } + // Log the final port configuration icebergPortStr := "disabled" if miniS3Options.portIceberg != nil && *miniS3Options.portIceberg > 0 { @@ -1616,6 +1630,9 @@ func startMiniAdminWithWorker(allServicesReady chan struct{}) { if miniProgressBoard != nil { miniProgressBoard.starting("Admin") } + // Snapshot the options, and with them the reserved listener, so a later + // in-process run cannot swap either out from under this goroutine. + adminOptions := miniAdminOptions done := trackMiniClient() go func() { defer done() @@ -1631,9 +1648,14 @@ func startMiniAdminWithWorker(allServicesReady chan struct{}) { lancePort = *miniS3Options.portLance } } - if err := startAdminServer(ctx, miniAdminOptions, *miniEnableAdminUI, icebergPort, lancePort, urlPrefix); err != nil { + if err := startAdminServer(ctx, adminOptions, *miniEnableAdminUI, icebergPort, lancePort, urlPrefix); err != nil { glog.Errorf("Admin server error: %v", err) } + // A no-op once the admin took it over and shut it down; it matters when + // startAdminServer bailed out before that. + if adminOptions.workerGrpcListener != nil { + adminOptions.workerGrpcListener.Close() + } }() // Wait for admin server's HTTP port to be ready before launching worker