mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-17 19:48:35 +02:00
s3: keep the gateway up when an auxiliary listener fails to bind (#11768)
* s3: keep the gateway up when the Iceberg or Lance listener fails to bind The optional catalog servers ran glog.Fatalf on a bind error, so an occupied port (e.g. the default Lance port 9101 taken by another process) killed the entire S3 server. Log the error and run without the auxiliary listener instead; the S3 API itself still fails hard on bind errors. * mini: only probe and advertise catalogs whose listeners bound With aux listener failures no longer fatal, the configured catalog port may answer as an unrelated process; mini's readiness probe accepts any HTTP response and would announce that foreign service as the catalog. Record the bind outcome on S3Options (final by the time the S3 port serves, since the aux binds happen first) and gate mini's readiness wait and admin UI advertisement on it. * s3: test that a failed aux listener bind is non-fatal Hold the Iceberg and Lance ports open and assert both starters return no server and leave the bound flags false, covering the new nonfatal path. * command: range the endpoint test table by index S3Options now holds atomic.Bool flags, so copying a test case trips go vet's lock-copy check. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
1 parent
cabaf9c38d
commit
5a0be275cc
4 files changed
+107
-11
No files matched your search
+21
-4
@@ -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
|
||||
}
|
||||
}
|
||||
|
||||
+25
-6
@@ -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
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
Reference in new issue
Block a user