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