diff --git a/weed/command/mini.go b/weed/command/mini.go index a44c55bb9..183e1961f 100644 --- a/weed/command/mini.go +++ b/weed/command/mini.go @@ -1627,13 +1627,30 @@ func startMiniServices(miniWhiteList []string, allServicesReady chan struct{}) { if miniProgressBoard != nil { miniProgressBoard.starting("Iceberg") } - waitForServiceReady("Iceberg", *miniS3Options.portIceberg, bindIp) + // Probe only when the catalog listener actually bound; otherwise the + // port may answer as an unrelated process and get advertised as the + // catalog. + if miniS3Options.icebergBound.Load() { + waitForServiceReady("Iceberg", *miniS3Options.portIceberg, bindIp) + } else { + if miniProgressBoard != nil { + miniProgressBoard.failed("Iceberg") + } + glog.Warningf("Iceberg catalog on port %d did not start; see S3 startup log", *miniS3Options.portIceberg) + } } if miniS3Options.portLance != nil && *miniS3Options.portLance > 0 { if miniProgressBoard != nil { miniProgressBoard.starting("Lance") } - waitForServiceReady("Lance", *miniS3Options.portLance, bindIp) + if miniS3Options.lanceBound.Load() { + waitForServiceReady("Lance", *miniS3Options.portLance, bindIp) + } else { + if miniProgressBoard != nil { + miniProgressBoard.failed("Lance") + } + glog.Warningf("Lance namespace server on port %d did not start; see S3 startup log", *miniS3Options.portLance) + } } } if *miniEnableWebDAV { @@ -1848,10 +1865,10 @@ func startMiniAdminWithWorker(allServicesReady chan struct{}) { // the admin UI would otherwise print an endpoint nothing is listening on. var icebergPort, lancePort int if miniEnableS3 != nil && *miniEnableS3 { - if miniS3Options.portIceberg != nil { + if miniS3Options.portIceberg != nil && *miniS3Options.portIceberg > 0 && miniS3Options.icebergBound.Load() { icebergPort = *miniS3Options.portIceberg } - if miniS3Options.portLance != nil { + if miniS3Options.portLance != nil && *miniS3Options.portLance > 0 && miniS3Options.lanceBound.Load() { lancePort = *miniS3Options.portLance } } diff --git a/weed/command/s3.go b/weed/command/s3.go index 7400266db..76308b24f 100644 --- a/weed/command/s3.go +++ b/weed/command/s3.go @@ -12,6 +12,7 @@ import ( "runtime" "strconv" "strings" + "sync/atomic" "time" "github.com/gorilla/mux" @@ -87,6 +88,14 @@ type S3Options struct { // 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 + + // icebergBound/lanceBound record whether the auxiliary catalog listeners + // bound their ports. The aux binds happen before the S3 listener serves, + // so once the S3 port answers these are final; mini reads them to probe + // and advertise only catalogs that are actually running instead of + // whatever process may hold the configured port. + icebergBound atomic.Bool + lanceBound atomic.Bool } func init() { @@ -399,12 +408,16 @@ func (s3opt *S3Options) startS3Server() bool { // Start Iceberg REST Catalog server if enabled if *s3opt.portIceberg > 0 { - httpServers = append(httpServers, s3opt.startIcebergServer(s3ApiServer)) + if httpS := s3opt.startIcebergServer(s3ApiServer); httpS != nil { + httpServers = append(httpServers, httpS) + } } // Start Lance Namespace server if enabled if s3opt.portLance != nil && *s3opt.portLance > 0 { - httpServers = append(httpServers, s3opt.startLanceServer(s3ApiServer)) + if httpS := s3opt.startLanceServer(s3ApiServer); httpS != nil { + httpServers = append(httpServers, httpS) + } } if runtime.GOOS != "windows" { @@ -601,8 +614,12 @@ func (s3opt *S3Options) startIcebergServer(s3ApiServer *s3api.S3ApiServer) *http icebergListener, icebergLocalListener, err := util.NewIpAndLocalListeners( *s3opt.bindIp, *s3opt.portIceberg, time.Duration(*s3opt.idleTimeout)*time.Second) if err != nil { - glog.Fatalf("Iceberg REST Catalog listener on %s error: %v", listenAddress, err) + // The catalog is auxiliary to the S3 gateway; a port conflict must not + // take the whole server down. + glog.Errorf("Iceberg REST Catalog listener on %s error: %v", listenAddress, err) + return nil } + s3opt.icebergBound.Store(true) glog.V(0).Infof("Start Iceberg REST Catalog Server at http://%s", listenAddress) @@ -617,7 +634,7 @@ func (s3opt *S3Options) startIcebergServer(s3ApiServer *s3api.S3ApiServer) *http } go func() { if err := httpS.Serve(icebergListener); err != nil && err != http.ErrServerClosed { - glog.Fatalf("Iceberg REST Catalog Server Fail to serve: %v", err) + glog.Errorf("Iceberg REST Catalog Server Fail to serve: %v", err) } }() return httpS @@ -643,8 +660,10 @@ func (s3opt *S3Options) startLanceServer(s3ApiServer *s3api.S3ApiServer) *http.S lanceListener, lanceLocalListener, err := util.NewIpAndLocalListeners( *s3opt.bindIp, *s3opt.portLance, time.Duration(*s3opt.idleTimeout)*time.Second) if err != nil { - glog.Fatalf("Lance Namespace listener on %s error: %v", listenAddress, err) + glog.Errorf("Lance Namespace listener on %s error: %v", listenAddress, err) + return nil } + s3opt.lanceBound.Store(true) glog.V(0).Infof("Start Lance Namespace Server at http://%s", listenAddress) @@ -658,7 +677,7 @@ func (s3opt *S3Options) startLanceServer(s3ApiServer *s3api.S3ApiServer) *http.S } go func() { if err := httpS.Serve(lanceListener); err != nil && err != http.ErrServerClosed { - glog.Fatalf("Lance Namespace Server Fail to serve: %v", err) + glog.Errorf("Lance Namespace Server Fail to serve: %v", err) } }() return httpS diff --git a/weed/command/s3_aux_listener_test.go b/weed/command/s3_aux_listener_test.go new file mode 100644 index 000000000..6ec5fac4b --- /dev/null +++ b/weed/command/s3_aux_listener_test.go @@ -0,0 +1,59 @@ +package command + +import ( + "net" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/s3api" +) + +func holdPort(t *testing.T) int { + t.Helper() + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + t.Cleanup(func() { l.Close() }) + return l.Addr().(*net.TCPAddr).Port +} + +// An occupied auxiliary catalog port must not take the S3 gateway down, must +// return no server for shutdown registration, and must not report the catalog +// as bound for mini's readiness checks. +func TestAuxListenerBindFailureIsNonFatal(t *testing.T) { + icebergPort := holdPort(t) + lancePort := holdPort(t) + + bind := "127.0.0.1" + empty := "" + zero := 0 + idle := 10 + s3opt := &S3Options{ + bindIp: &bind, + ip: &bind, + port: &zero, + portHttps: &zero, + portIceberg: &icebergPort, + portLance: &lancePort, + idleTimeout: &idle, + icebergCredentialRole: &empty, + externalUrl: &empty, + tlsPrivateKey: &empty, + } + s3a := &s3api.S3ApiServer{} + + if s := s3opt.startIcebergServer(s3a); s != nil { + s.Close() + t.Error("startIcebergServer on an occupied port should return nil") + } + if s3opt.icebergBound.Load() { + t.Error("icebergBound must stay false when the listener failed to bind") + } + if s := s3opt.startLanceServer(s3a); s != nil { + s.Close() + t.Error("startLanceServer on an occupied port should return nil") + } + if s3opt.lanceBound.Load() { + t.Error("lanceBound must stay false when the listener failed to bind") + } +} diff --git a/weed/command/s3_iceberg_endpoint_test.go b/weed/command/s3_iceberg_endpoint_test.go index 14608aeb7..337a487b8 100644 --- a/weed/command/s3_iceberg_endpoint_test.go +++ b/weed/command/s3_iceberg_endpoint_test.go @@ -92,7 +92,8 @@ func TestDeriveS3AdvertisedEndpoint(t *testing.T) { }, } - for _, tc := range tests { + for i := range tests { + tc := &tests[i] t.Run(tc.name, func(t *testing.T) { got := tc.opt.deriveS3AdvertisedEndpoint() if got != tc.want {