mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-26 06:50:45 +01:00
command: no extra localhost listener when -ip.bind is specified (#12055)
* util: add listenOnLocalhost parameter to NewIpAndLocalListeners Nil keeps the current behavior of also binding localhost when the host is a specific address; all callers pass nil for now. * command: add -ip.bind.localhost flag to opt out of the extra localhost listener Binding a command to a specific -ip.bind address also opened a second listener on localhost for every port, with no way to turn it off. Each command that calls NewIpAndLocalListeners (master, master.follower, filer, s3, iam, volume, sftp, mq.broker, mq.agent) now accepts -ip.bind.localhost=false to bind only the requested address. Embedded services get prefixed flags (s3.ip.bind.localhost, sftp.ip.bind.localhost, iam.ip.bind.localhost), and the command-level flag also disables them so a single switch turns off every extra listener the command opens. * mini: only require all-interface port availability while the loopback listener is on With -ip.bind.localhost=false each service binds just -ip.bind, so a port held on another address no longer collides; the all-interface precheck would reject or relocate usable ports. WebDAV, the Admin UI, and the Admin worker gRPC (which binds -admin.worker.ip, not -ip.bind) never open a loopback companion listener, so their checks now use the address they actually bind. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * mq.broker: bind the gRPC listener to -ip.bind instead of all interfaces The listener used an empty host so -ip was ignored for binding and the localhost companion was never created, leaving -ip.bind.localhost with nothing to control. Give the broker the same -ip/-ip.bind pair the other servers use; embedded mode binds to -ip.bind like its peers. * server: add -sftp.ip.bind.localhost and -iam.ip.bind.localhost switches weed filer already exposes the prefixed opt-outs for its embedded sftp and iam; give weed server the same pair so each extra listener can be disabled individually while -ip.bind.localhost=false still turns them all off. * mq.broker: keep the default all-interface bind unless localhost is off An unset -ip.bind used to mean the standalone broker listened on every interface; defaulting it to -ip silently cut off multi-homed clients. Only fall back to -ip when -ip.bind.localhost=false, where a specific bind is needed for the opt-out to take effect. * mini: validate the Admin HTTP port on the address it will bind The port check always used -ip.bind while startMiniAdminWithWorker falls back to loopback when Admin has no credential, so a loopback-held port passed validation and killed the admin at startup. * mq.broker: keep the all-interface default bind when -ip.bind is unset -ip only becomes the fallback once -ip.bind.localhost=false needs a specific address for the listener. Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * mini: validate the Admin HTTP port on its effective bind IP When the Admin listener falls back to loopback (no credential or mTLS configured), the pre-flight port check now probes that same address instead of -ip.bind. Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * util: keep NewIpAndLocalListeners source-compatible for 3-arg callers * command: drop the -ip.bind.localhost flag; an explicit -ip.bind drops the extra listener Instead of an opt-out flag, the loopback companion listener now simply follows whether the bind address was chosen: it is created only while -ip.bind is unset and the bind address falls back to -ip. Passing -ip.bind=<addr> binds each port to exactly that address. Each service options struct carries a listenOnLocalhost *bool that embedders (weed server, weed mini, embedded s3/sftp/iam in weed filer) set from the parent's -ip.bind state, captured before the bind flag is repointed or filled in by the -ip fallback; standalone commands derive it from their own flag before the fallback runs. iam and mq.agent gain a real -ip.bind flag so they have a way to pin a single address. mq.broker keeps its new -ip.bind flag but drops the -ip.bind.localhost fallback hack: empty still listens on all interfaces. * server: default embedded mq.agent brokers to the broker's bind address With an explicit -ip.bind the embedded broker no longer listens on localhost, so an unset -mq.agent.brokers defaulting to localhost:17777 could not reach it. Default to <serverBindIp>:<mq.broker.port> (the address the embedded broker binds), falling back to localhost when the bind address is a wildcard. An explicit -mq.agent.brokers still wins. Also fix the NewIpAndLocalListeners skip message to say localhost. * server: bind embedded mq.agent to serverBindIp The embedded agent's ipBind was never set, so startQueueAgent fell back to the advertised -ip address and an explicit -ip.bind did not pin the agent listener. --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
1 parent
fefac028ec
commit
88b91cccba
13 files changed
+307
-119
No files matched your search
+18
-3
@@ -64,6 +64,7 @@ type FilerOptions struct {
|
||||
mastersString *string
|
||||
ip *string
|
||||
bindIp *string
|
||||
listenOnLocalhost *bool
|
||||
port *int
|
||||
portGrpc *int
|
||||
publicPort *int
|
||||
@@ -282,12 +283,17 @@ func runFiler(cmd *Command, args []string) bool {
|
||||
}
|
||||
go stats_collect.StartMetricsServer(*f.metricsHttpIp, *f.metricsHttpPort)
|
||||
|
||||
// Embedded services keep their loopback companion listener only while
|
||||
// -ip.bind is unset and the bind address falls back to -ip. This must be
|
||||
// captured before startFiler fills *f.bindIp in place.
|
||||
filerBindIpUnset := *f.bindIp == ""
|
||||
filerAddress := pb.NewServerAddress(*f.ip, *f.port, *f.portGrpc).String()
|
||||
startDelay := time.Duration(2)
|
||||
if *filerStartS3 {
|
||||
filerS3Options.filer = &filerAddress
|
||||
filerS3Options.ip = f.ip
|
||||
if *filerS3Options.bindIp == "" {
|
||||
filerS3Options.listenOnLocalhost = &filerBindIpUnset
|
||||
filerS3Options.bindIp = f.bindIp
|
||||
}
|
||||
filerS3Options.localFilerSocket = f.localSocket
|
||||
@@ -320,6 +326,7 @@ func runFiler(cmd *Command, args []string) bool {
|
||||
if *filerStartIam {
|
||||
filerIamOptions.filer = &filerAddress
|
||||
filerIamOptions.masters = f.mastersString
|
||||
filerIamOptions.listenOnLocalhost = &filerBindIpUnset
|
||||
go func(delay time.Duration) {
|
||||
time.Sleep(delay * time.Second)
|
||||
filerIamOptions.startIamServer()
|
||||
@@ -330,6 +337,7 @@ func runFiler(cmd *Command, args []string) bool {
|
||||
if *filerStartSftp {
|
||||
filerSftpOptions.filer = &filerAddress
|
||||
if *filerSftpOptions.bindIp == "" {
|
||||
filerSftpOptions.listenOnLocalhost = &filerBindIpUnset
|
||||
filerSftpOptions.bindIp = f.bindIp
|
||||
}
|
||||
go func(delay time.Duration) {
|
||||
@@ -356,6 +364,13 @@ func (fo *FilerOptions) startFiler() {
|
||||
if *fo.portGrpc == 0 {
|
||||
*fo.portGrpc = 10000 + *fo.port
|
||||
}
|
||||
// The loopback companion listener is only kept while -ip.bind is unset
|
||||
// and the bind address falls back to -ip. An explicit -ip.bind pins each
|
||||
// port to exactly that address.
|
||||
listenOnLocalhost := *fo.bindIp == ""
|
||||
if fo.listenOnLocalhost != nil {
|
||||
listenOnLocalhost = *fo.listenOnLocalhost
|
||||
}
|
||||
if *fo.bindIp == "" {
|
||||
*fo.bindIp = *fo.ip
|
||||
}
|
||||
@@ -433,7 +448,7 @@ func (fo *FilerOptions) startFiler() {
|
||||
if *fo.publicPort != 0 {
|
||||
publicListeningAddress := util.JoinHostPort(*fo.bindIp, *fo.publicPort)
|
||||
glog.V(0).Infoln("Start Seaweed filer server", version.Version(), "public at", publicListeningAddress)
|
||||
publicListener, localPublicListener, e := util.NewIpAndLocalListeners(*fo.bindIp, *fo.publicPort, 0)
|
||||
publicListener, localPublicListener, e := util.NewIpAndLocalListeners(*fo.bindIp, *fo.publicPort, 0, listenOnLocalhost)
|
||||
if e != nil {
|
||||
glog.Fatalf("Filer server public listener error on port %d:%v", *fo.publicPort, e)
|
||||
}
|
||||
@@ -459,7 +474,7 @@ func (fo *FilerOptions) startFiler() {
|
||||
glog.V(0).Infof("Start Seaweed Filer %s at %s:%d", version.Version(), *fo.ip, *fo.port)
|
||||
filerListener, filerLocalListener, e := util.NewIpAndLocalListeners(
|
||||
*fo.bindIp, *fo.port,
|
||||
time.Duration(10)*time.Second,
|
||||
time.Duration(10)*time.Second, listenOnLocalhost,
|
||||
)
|
||||
if e != nil {
|
||||
glog.Fatalf("Filer listener error: %v", e)
|
||||
@@ -467,7 +482,7 @@ func (fo *FilerOptions) startFiler() {
|
||||
|
||||
// starting grpc server
|
||||
grpcPort := *fo.portGrpc
|
||||
grpcL, grpcLocalL, err := util.NewIpAndLocalListeners(*fo.bindIp, grpcPort, 0)
|
||||
grpcL, grpcLocalL, err := util.NewIpAndLocalListeners(*fo.bindIp, grpcPort, 0, listenOnLocalhost)
|
||||
if err != nil {
|
||||
glog.Fatalf("failed to listen on grpc port %d: %v", grpcPort, err)
|
||||
}
|
||||
|
||||
+16
-1
@@ -31,6 +31,8 @@ type IamOptions struct {
|
||||
filer *string
|
||||
masters *string
|
||||
ip *string
|
||||
ipBind *string
|
||||
listenOnLocalhost *bool
|
||||
port *int
|
||||
selfServiceRequiresGrant *bool
|
||||
}
|
||||
@@ -40,6 +42,7 @@ func init() {
|
||||
iamStandaloneOptions.filer = cmdIam.Flag.String("filer", "localhost:8888", "comma-separated filer server addresses for high availability")
|
||||
iamStandaloneOptions.masters = cmdIam.Flag.String("master", "localhost:9333", "comma-separated master servers")
|
||||
iamStandaloneOptions.ip = cmdIam.Flag.String("ip", util.DetectedHostAddress(), "iam server http listen ip address")
|
||||
iamStandaloneOptions.ipBind = cmdIam.Flag.String("ip.bind", "", "ip address to bind to. If empty, default to same as -ip option.")
|
||||
iamStandaloneOptions.port = cmdIam.Flag.Int("port", 8111, "iam server http listen port")
|
||||
iamStandaloneOptions.selfServiceRequiresGrant = cmdIam.Flag.Bool("selfServiceRequiresGrant", false, iamSelfServiceRequiresGrantUsage)
|
||||
}
|
||||
@@ -119,8 +122,20 @@ func (iamopt *IamOptions) startIamServer() bool {
|
||||
iamApiServer.Shutdown()
|
||||
})
|
||||
|
||||
// The loopback companion listener is only kept while -ip.bind is unset and
|
||||
// the bind address falls back to -ip. An explicit -ip.bind pins the port to
|
||||
// exactly that address.
|
||||
bindIp := *iamopt.ip
|
||||
listenOnLocalhost := iamopt.ipBind == nil || *iamopt.ipBind == ""
|
||||
if iamopt.ipBind != nil && *iamopt.ipBind != "" {
|
||||
bindIp = *iamopt.ipBind
|
||||
}
|
||||
if iamopt.listenOnLocalhost != nil {
|
||||
listenOnLocalhost = *iamopt.listenOnLocalhost
|
||||
}
|
||||
|
||||
listenAddress := fmt.Sprintf(":%d", *iamopt.port)
|
||||
iamApiListener, iamApiLocalListener, err := util.NewIpAndLocalListeners(*iamopt.ip, *iamopt.port, time.Duration(10)*time.Second)
|
||||
iamApiListener, iamApiLocalListener, err := util.NewIpAndLocalListeners(bindIp, *iamopt.port, time.Duration(10)*time.Second, listenOnLocalhost)
|
||||
if err != nil {
|
||||
glog.Fatalf("IAM API Server listener on %s error: %v", listenAddress, err)
|
||||
}
|
||||
|
||||
+10
-2
@@ -50,6 +50,7 @@ type MasterOptions struct {
|
||||
portGrpc *int
|
||||
ip *string
|
||||
ipBind *string
|
||||
listenOnLocalhost *bool
|
||||
metaFolder *string
|
||||
peers *string
|
||||
mastersDeprecated *string // deprecated, for backward compatibility in master.follower
|
||||
@@ -191,6 +192,13 @@ func startMaster(masterOption MasterOptions, masterWhiteList []string) {
|
||||
if *masterOption.portGrpc == 0 {
|
||||
*masterOption.portGrpc = 10000 + *masterOption.port
|
||||
}
|
||||
// The loopback companion listener is only kept while -ip.bind is unset
|
||||
// and the bind address falls back to -ip. An explicit -ip.bind pins each
|
||||
// port to exactly that address.
|
||||
listenOnLocalhost := *masterOption.ipBind == ""
|
||||
if masterOption.listenOnLocalhost != nil {
|
||||
listenOnLocalhost = *masterOption.listenOnLocalhost
|
||||
}
|
||||
if *masterOption.ipBind == "" {
|
||||
*masterOption.ipBind = *masterOption.ip
|
||||
}
|
||||
@@ -207,7 +215,7 @@ func startMaster(masterOption MasterOptions, masterWhiteList []string) {
|
||||
ms := weed_server.NewMasterServer(r, masterOption.toMasterOption(masterWhiteList), masterPeers)
|
||||
listeningAddress := util.JoinHostPort(*masterOption.ipBind, *masterOption.port)
|
||||
glog.V(0).Infof("Start Seaweed Master %s at %s", version.Version(), listeningAddress)
|
||||
masterListener, masterLocalListener, e := util.NewIpAndLocalListeners(*masterOption.ipBind, *masterOption.port, 0)
|
||||
masterListener, masterLocalListener, e := util.NewIpAndLocalListeners(*masterOption.ipBind, *masterOption.port, 0, listenOnLocalhost)
|
||||
if e != nil {
|
||||
glog.Fatalf("Master startup error: %v", e)
|
||||
}
|
||||
@@ -264,7 +272,7 @@ func startMaster(masterOption MasterOptions, masterWhiteList []string) {
|
||||
}
|
||||
// starting grpc server
|
||||
grpcPort := *masterOption.portGrpc
|
||||
grpcL, grpcLocalL, err := util.NewIpAndLocalListeners(*masterOption.ipBind, grpcPort, 0)
|
||||
grpcL, grpcLocalL, err := util.NewIpAndLocalListeners(*masterOption.ipBind, grpcPort, 0, listenOnLocalhost)
|
||||
if err != nil {
|
||||
glog.Fatalf("master failed to listen on grpc port %d: %v", grpcPort, err)
|
||||
}
|
||||
|
||||
@@ -129,6 +129,10 @@ func startMasterFollower(masterOptions MasterOptions) {
|
||||
option := masterOptions.toMasterOption(nil)
|
||||
option.IsFollower = true
|
||||
|
||||
listenOnLocalhost := *masterOptions.ipBind == ""
|
||||
if masterOptions.listenOnLocalhost != nil {
|
||||
listenOnLocalhost = *masterOptions.listenOnLocalhost
|
||||
}
|
||||
if *masterOptions.ipBind == "" {
|
||||
*masterOptions.ipBind = *masterOptions.ip
|
||||
}
|
||||
@@ -137,14 +141,14 @@ func startMasterFollower(masterOptions MasterOptions) {
|
||||
ms := weed_server.NewMasterServer(r, option, masters)
|
||||
listeningAddress := util.JoinHostPort(*masterOptions.ipBind, *masterOptions.port)
|
||||
glog.V(0).Infof("Start Seaweed Master %s at %s", version.Version(), listeningAddress)
|
||||
masterListener, masterLocalListener, e := util.NewIpAndLocalListeners(*masterOptions.ipBind, *masterOptions.port, 0)
|
||||
masterListener, masterLocalListener, e := util.NewIpAndLocalListeners(*masterOptions.ipBind, *masterOptions.port, 0, listenOnLocalhost)
|
||||
if e != nil {
|
||||
glog.Fatalf("Master startup error: %v", e)
|
||||
}
|
||||
|
||||
// starting grpc server
|
||||
grpcPort := *masterOptions.portGrpc
|
||||
grpcL, grpcLocalL, err := util.NewIpAndLocalListeners(*masterOptions.ipBind, grpcPort, 0)
|
||||
grpcL, grpcLocalL, err := util.NewIpAndLocalListeners(*masterOptions.ipBind, grpcPort, 0, listenOnLocalhost)
|
||||
if err != nil {
|
||||
glog.Fatalf("master failed to listen on grpc port %d: %v", grpcPort, err)
|
||||
}
|
||||
|
||||
+111
-65
@@ -821,10 +821,20 @@ func isPortAvailable(port int) bool {
|
||||
return true
|
||||
}
|
||||
|
||||
// isPortUsableOnIP reports whether port can be bound on ip. While mini also
|
||||
// opens a loopback listener, the port must additionally be free on every
|
||||
// interface so that listener cannot collide with a holder on another address.
|
||||
func isPortUsableOnIP(ip string, port int, listenOnLocalhost bool) bool {
|
||||
if !isPortOpenOnIP(ip, port) {
|
||||
return false
|
||||
}
|
||||
return !listenOnLocalhost || isPortAvailable(port)
|
||||
}
|
||||
|
||||
// findAvailablePortOnIP finds the next available port on a specific IP starting from the given port
|
||||
// It skips any ports that are in the reservedPorts map (for gRPC port collision avoidance)
|
||||
// It returns the first available port found within maxAttempts, or 0 if none found
|
||||
func findAvailablePortOnIP(ip string, startPort int, maxAttempts int, reservedPorts map[int]bool) int {
|
||||
func findAvailablePortOnIP(ip string, startPort int, maxAttempts int, reservedPorts map[int]bool, listenOnLocalhost bool) int {
|
||||
for i := 0; i < maxAttempts; i++ {
|
||||
port := startPort + i
|
||||
if port > 65535 {
|
||||
@@ -836,7 +846,7 @@ func findAvailablePortOnIP(ip string, startPort int, maxAttempts int, reservedPo
|
||||
continue
|
||||
}
|
||||
// Check on both the specific IP and on all interfaces for maximum reliability
|
||||
if isPortOpenOnIP(ip, port) && isPortAvailable(port) {
|
||||
if isPortUsableOnIP(ip, port, listenOnLocalhost) {
|
||||
return port
|
||||
}
|
||||
}
|
||||
@@ -847,7 +857,7 @@ func findAvailablePortOnIP(ip string, startPort int, maxAttempts int, reservedPo
|
||||
// ensurePortAvailableOnIP ensures a port pointer points to an available port on a specific IP
|
||||
// If the port is not available, it finds the next available port and updates the pointer
|
||||
// The reservedPorts map contains ports that should not be allocated (for gRPC collision avoidance)
|
||||
func ensurePortAvailableOnIP(portPtr *int, serviceName string, ip string, reservedPorts map[int]bool, flagName string) error {
|
||||
func ensurePortAvailableOnIP(portPtr *int, serviceName string, ip string, reservedPorts map[int]bool, flagName string, listenOnLocalhost bool) error {
|
||||
// Check if this port was explicitly specified by the user (from CLI or env, before the config file was applied)
|
||||
isExplicitPort := explicitFlags[flagName]
|
||||
|
||||
@@ -863,7 +873,7 @@ func ensurePortAvailableOnIP(portPtr *int, serviceName string, ip string, reserv
|
||||
return fmt.Errorf("port %d for %s (specified by flag %s) is reserved for gRPC calculation and cannot be used", original, serviceName, flagName)
|
||||
}
|
||||
glog.Warningf("Port %d for %s is reserved for gRPC calculation, finding alternative...", original, serviceName)
|
||||
newPort := findAvailablePortOnIP(ip, original+1, 100, reservedPorts)
|
||||
newPort := findAvailablePortOnIP(ip, original+1, 100, reservedPorts, listenOnLocalhost)
|
||||
if newPort == 0 {
|
||||
glog.Errorf("Could not find available port for %s starting from %d, will use original %d and fail on binding", serviceName, original+1, original)
|
||||
} else {
|
||||
@@ -874,14 +884,14 @@ func ensurePortAvailableOnIP(portPtr *int, serviceName string, ip string, reserv
|
||||
}
|
||||
|
||||
// Check on both the specific IP and on all interfaces (0.0.0.0) for maximum reliability
|
||||
if !isPortOpenOnIP(ip, original) || !isPortAvailable(original) {
|
||||
if !isPortUsableOnIP(ip, original, listenOnLocalhost) {
|
||||
// If explicitly specified, fail immediately with the originally requested port
|
||||
if isExplicitPort {
|
||||
return fmt.Errorf("port %d for %s (specified by flag %s) is not available on %s and cannot be used", original, serviceName, flagName, ip)
|
||||
}
|
||||
// For default ports, try to find an alternative
|
||||
glog.Warningf("Port %d for %s is not available on %s, finding alternative port...", original, serviceName, ip)
|
||||
newPort := findAvailablePortOnIP(ip, original+1, 100, reservedPorts)
|
||||
newPort := findAvailablePortOnIP(ip, original+1, 100, reservedPorts, listenOnLocalhost)
|
||||
if newPort == 0 {
|
||||
glog.Errorf("Could not find available port for %s starting from %d, will use original %d and fail on binding", serviceName, original+1, original)
|
||||
} else {
|
||||
@@ -897,53 +907,61 @@ func ensurePortAvailableOnIP(portPtr *int, serviceName string, ip string, reserv
|
||||
// ensureAllPortsAvailableOnIP ensures all mini service ports are available on a specific IP
|
||||
// Returns an error if an explicitly specified port is unavailable.
|
||||
// This should be called before starting any services
|
||||
func ensureAllPortsAvailableOnIP(bindIp string) error {
|
||||
func ensureAllPortsAvailableOnIP(bindIp string, listenOnLocalhost bool) error {
|
||||
portConfigs := []struct {
|
||||
port *int
|
||||
name string
|
||||
flagName string
|
||||
grpcPtr *int
|
||||
port *int
|
||||
name string
|
||||
flagName string
|
||||
grpcPtr *int
|
||||
localhost bool
|
||||
}{
|
||||
{miniMasterOptions.port, "Master", "master.port", miniMasterOptions.portGrpc},
|
||||
{miniFilerOptions.port, "Filer", "filer.port", miniFilerOptions.portGrpc},
|
||||
{miniOptions.v.port, "Volume", "volume.port", miniOptions.v.portGrpc},
|
||||
{miniMasterOptions.port, "Master", "master.port", miniMasterOptions.portGrpc, listenOnLocalhost},
|
||||
{miniFilerOptions.port, "Filer", "filer.port", miniFilerOptions.portGrpc, listenOnLocalhost},
|
||||
{miniOptions.v.port, "Volume", "volume.port", miniOptions.v.portGrpc, listenOnLocalhost},
|
||||
}
|
||||
if *miniEnableS3 {
|
||||
portConfigs = append(portConfigs, struct {
|
||||
port *int
|
||||
name string
|
||||
flagName string
|
||||
grpcPtr *int
|
||||
}{miniS3Options.port, "S3", "s3.port", miniS3Options.portGrpc})
|
||||
port *int
|
||||
name string
|
||||
flagName string
|
||||
grpcPtr *int
|
||||
localhost bool
|
||||
}{miniS3Options.port, "S3", "s3.port", miniS3Options.portGrpc, listenOnLocalhost})
|
||||
if miniS3Options.portIceberg != nil && *miniS3Options.portIceberg > 0 {
|
||||
portConfigs = append(portConfigs, struct {
|
||||
port *int
|
||||
name string
|
||||
flagName string
|
||||
grpcPtr *int
|
||||
}{miniS3Options.portIceberg, "Iceberg", "s3.port.iceberg", nil})
|
||||
port *int
|
||||
name string
|
||||
flagName string
|
||||
grpcPtr *int
|
||||
localhost bool
|
||||
}{miniS3Options.portIceberg, "Iceberg", "s3.port.iceberg", nil, listenOnLocalhost})
|
||||
}
|
||||
if miniS3Options.portLance != nil && *miniS3Options.portLance > 0 {
|
||||
portConfigs = append(portConfigs, struct {
|
||||
port *int
|
||||
name string
|
||||
flagName string
|
||||
grpcPtr *int
|
||||
}{miniS3Options.portLance, "Lance", "s3.port.lance", nil})
|
||||
port *int
|
||||
name string
|
||||
flagName string
|
||||
grpcPtr *int
|
||||
localhost bool
|
||||
}{miniS3Options.portLance, "Lance", "s3.port.lance", nil, listenOnLocalhost})
|
||||
}
|
||||
}
|
||||
// WebDAV and the Admin UI bind only -ip.bind; they never open a loopback
|
||||
// companion listener, so ports held elsewhere cannot collide with them.
|
||||
portConfigs = append(portConfigs, struct {
|
||||
port *int
|
||||
name string
|
||||
flagName string
|
||||
grpcPtr *int
|
||||
}{miniWebDavOptions.port, "WebDAV", "webdav.port", nil},
|
||||
port *int
|
||||
name string
|
||||
flagName string
|
||||
grpcPtr *int
|
||||
localhost bool
|
||||
}{miniWebDavOptions.port, "WebDAV", "webdav.port", nil, false},
|
||||
struct {
|
||||
port *int
|
||||
name string
|
||||
flagName string
|
||||
grpcPtr *int
|
||||
}{miniAdminOptions.port, "Admin", "admin.port", miniAdminOptions.grpcPort})
|
||||
port *int
|
||||
name string
|
||||
flagName string
|
||||
grpcPtr *int
|
||||
localhost bool
|
||||
}{miniAdminOptions.port, "Admin", "admin.port", miniAdminOptions.grpcPort, false})
|
||||
|
||||
// First, reserve all gRPC ports that will be calculated to prevent HTTP port allocation from using them
|
||||
// This prevents collisions like: HTTP port moves to X, then gRPC port is calculated as Y where Y == X
|
||||
@@ -956,13 +974,20 @@ func ensureAllPortsAvailableOnIP(bindIp string) error {
|
||||
}
|
||||
}
|
||||
|
||||
applyMiniAdminCredentialFallback(&miniAdminOptions)
|
||||
adminBindIP := miniAdminEffectiveBindIP(bindIp)
|
||||
|
||||
// Check all HTTP ports sequentially to avoid race conditions
|
||||
// Each port check and allocation must complete before the next one starts
|
||||
// to prevent multiple goroutines from claiming the same available port
|
||||
// Also avoid allocating ports that are reserved for gRPC calculation
|
||||
for _, config := range portConfigs {
|
||||
checkIP := bindIp
|
||||
if config.name == "Admin" {
|
||||
checkIP = adminBindIP
|
||||
}
|
||||
original := *config.port
|
||||
if err := ensurePortAvailableOnIP(config.port, config.name, bindIp, reservedPorts, config.flagName); err != nil {
|
||||
if err := ensurePortAvailableOnIP(config.port, config.name, checkIP, reservedPorts, config.flagName, config.localhost); err != nil {
|
||||
return err
|
||||
}
|
||||
// If port was changed, update the reserved gRPC ports mapping
|
||||
@@ -975,7 +1000,7 @@ func ensureAllPortsAvailableOnIP(bindIp string) error {
|
||||
// Initialize all gRPC ports before services start
|
||||
// This ensures they won't be recalculated and cause conflicts
|
||||
// All gRPC port handling (calculation, validation, and assignment) is performed exclusively in initializeGrpcPortsOnIP
|
||||
allocatedPorts := initializeGrpcPortsOnIP(bindIp)
|
||||
allocatedPorts := initializeGrpcPortsOnIP(bindIp, listenOnLocalhost)
|
||||
|
||||
// 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
|
||||
@@ -1054,7 +1079,7 @@ func listenMiniAdminWorker(bindIP string, port int) (net.Listener, error) {
|
||||
// This must be called after HTTP ports are finalized and before services start
|
||||
// Returns the set of ports allocated to mini services, for callers that must
|
||||
// keep optional listeners from stealing them.
|
||||
func initializeGrpcPortsOnIP(bindIp string) map[int]bool {
|
||||
func initializeGrpcPortsOnIP(bindIp string, listenOnLocalhost bool) map[int]bool {
|
||||
// Track all ports allocated (both HTTP and gRPC) to prevent collisions.
|
||||
// We must reserve HTTP ports so that gRPC fallback allocation never picks
|
||||
// a port already assigned to an HTTP service (which hasn't bound yet).
|
||||
@@ -1077,26 +1102,34 @@ func initializeGrpcPortsOnIP(bindIp string) map[int]bool {
|
||||
}
|
||||
|
||||
grpcConfigs := []struct {
|
||||
httpPort *int
|
||||
grpcPort *int
|
||||
name string
|
||||
httpPort *int
|
||||
grpcPort *int
|
||||
name string
|
||||
ip string
|
||||
localhost bool
|
||||
}{
|
||||
{miniMasterOptions.port, miniMasterOptions.portGrpc, "Master"},
|
||||
{miniFilerOptions.port, miniFilerOptions.portGrpc, "Filer"},
|
||||
{miniOptions.v.port, miniOptions.v.portGrpc, "Volume"},
|
||||
{miniMasterOptions.port, miniMasterOptions.portGrpc, "Master", bindIp, listenOnLocalhost},
|
||||
{miniFilerOptions.port, miniFilerOptions.portGrpc, "Filer", bindIp, listenOnLocalhost},
|
||||
{miniOptions.v.port, miniOptions.v.portGrpc, "Volume", bindIp, listenOnLocalhost},
|
||||
}
|
||||
if *miniEnableS3 {
|
||||
grpcConfigs = append(grpcConfigs, struct {
|
||||
httpPort *int
|
||||
grpcPort *int
|
||||
name string
|
||||
}{miniS3Options.port, miniS3Options.portGrpc, "S3"})
|
||||
httpPort *int
|
||||
grpcPort *int
|
||||
name string
|
||||
ip string
|
||||
localhost bool
|
||||
}{miniS3Options.port, miniS3Options.portGrpc, "S3", bindIp, listenOnLocalhost})
|
||||
}
|
||||
// The Admin worker gRPC binds workerGrpcBindIp (loopback by default), not
|
||||
// -ip.bind, and never opens a companion listener.
|
||||
grpcConfigs = append(grpcConfigs, struct {
|
||||
httpPort *int
|
||||
grpcPort *int
|
||||
name string
|
||||
}{miniAdminOptions.port, miniAdminOptions.grpcPort, "Admin"})
|
||||
httpPort *int
|
||||
grpcPort *int
|
||||
name string
|
||||
ip string
|
||||
localhost bool
|
||||
}{miniAdminOptions.port, miniAdminOptions.grpcPort, "Admin", miniAdminOptions.workerGrpcBindIp, false})
|
||||
|
||||
for _, config := range grpcConfigs {
|
||||
if config.grpcPort == nil {
|
||||
@@ -1110,10 +1143,10 @@ func initializeGrpcPortsOnIP(bindIp string) map[int]bool {
|
||||
|
||||
// Verify the gRPC port is available (whether calculated or explicitly set)
|
||||
// Check on both specific IP and all interfaces, and check against already allocated ports
|
||||
if !isPortOpenOnIP(bindIp, *config.grpcPort) || !isPortAvailable(*config.grpcPort) || allocatedPorts[*config.grpcPort] {
|
||||
if !isPortUsableOnIP(config.ip, *config.grpcPort, config.localhost) || allocatedPorts[*config.grpcPort] {
|
||||
glog.Warningf("gRPC port %d for %s is not available, finding alternative...", *config.grpcPort, config.name)
|
||||
originalPort := *config.grpcPort
|
||||
newPort := findAvailablePortOnIP(bindIp, originalPort+1, 100, allocatedPorts)
|
||||
newPort := findAvailablePortOnIP(config.ip, originalPort+1, 100, allocatedPorts, config.localhost)
|
||||
if newPort == 0 {
|
||||
glog.Errorf("Could not find available gRPC port for %s starting from %d, will use %d and fail on binding", config.name, originalPort+1, originalPort)
|
||||
} else {
|
||||
@@ -1404,8 +1437,14 @@ func runMini(cmd *Command, args []string) bool {
|
||||
miniAdminOptions.workerGrpcBindIp = "127.0.0.1"
|
||||
}
|
||||
|
||||
// Services keep the loopback companion listener only while -ip.bind is
|
||||
// unset and the bind address falls back to -ip. While it is on, each
|
||||
// service port must be free on every interface, not just bindIp. This
|
||||
// must be captured before *miniBindIp is filled in below.
|
||||
listenOnLocalhost := *miniBindIp == "" && !util.IsLoopbackOrWildcardAddress(bindIp)
|
||||
|
||||
// Ensure all ports are available, find alternatives if needed
|
||||
if err := ensureAllPortsAvailableOnIP(bindIp); err != nil {
|
||||
if err := ensureAllPortsAvailableOnIP(bindIp, listenOnLocalhost); err != nil {
|
||||
glog.Errorf("Port allocation failed: %v", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
@@ -1430,14 +1469,18 @@ func runMini(cmd *Command, args []string) bool {
|
||||
// ip address
|
||||
miniMasterOptions.ip = miniIp
|
||||
miniMasterOptions.ipBind = miniBindIp
|
||||
miniMasterOptions.listenOnLocalhost = &listenOnLocalhost
|
||||
miniFilerOptions.masters = pb.ServerAddresses(actualPeersForComponents).ToServiceDiscovery()
|
||||
miniFilerOptions.ip = miniIp
|
||||
miniFilerOptions.bindIp = miniBindIp
|
||||
miniFilerOptions.listenOnLocalhost = &listenOnLocalhost
|
||||
miniS3Options.ip = miniIp
|
||||
miniS3Options.bindIp = miniBindIp
|
||||
miniS3Options.listenOnLocalhost = &listenOnLocalhost
|
||||
miniWebDavOptions.ipBind = miniBindIp
|
||||
miniOptions.v.ip = miniIp
|
||||
miniOptions.v.bindIp = miniBindIp
|
||||
miniOptions.v.listenOnLocalhost = &listenOnLocalhost
|
||||
miniOptions.v.masters = pb.ServerAddresses(actualPeersForComponents).ToAddresses()
|
||||
miniOptions.v.idleConnectionTimeout = miniTimeout
|
||||
// Admin server binds to the same address as the other mini services.
|
||||
@@ -1776,6 +1819,14 @@ func miniAdminBindIP(requestedIP string, passwordConfigured, mtlsConfigured, all
|
||||
return "127.0.0.1"
|
||||
}
|
||||
|
||||
func miniAdminEffectiveBindIP(requestedIP string) string {
|
||||
hasMTLS := util.GetViper().GetString("https.admin.key") != "" &&
|
||||
util.GetViper().GetString("https.admin.ca") != ""
|
||||
allowInsecure := miniAdminOptions.allowInsecureBind != nil &&
|
||||
*miniAdminOptions.allowInsecureBind
|
||||
return miniAdminBindIP(requestedIP, *miniAdminOptions.adminPassword != "", hasMTLS, allowInsecure)
|
||||
}
|
||||
|
||||
// miniAdminWorkerAddress encodes the finalized HTTP and gRPC ports in the
|
||||
// server-address format understood by pb.ServerToGrpcAddress.
|
||||
func miniAdminWorkerAddress(bindIP string, httpPort, grpcPort int) string {
|
||||
@@ -1833,16 +1884,11 @@ func startMiniAdminWithWorker(allServicesReady chan struct{}) {
|
||||
applyMiniAdminCredentialFallback(&miniAdminOptions)
|
||||
|
||||
requestedBindIP := getBindIp()
|
||||
bindIP := miniAdminEffectiveBindIP(requestedBindIP)
|
||||
hasMTLS := util.GetViper().GetString("https.admin.key") != "" &&
|
||||
util.GetViper().GetString("https.admin.ca") != ""
|
||||
allowInsecure := miniAdminOptions.allowInsecureBind != nil &&
|
||||
*miniAdminOptions.allowInsecureBind
|
||||
bindIP := miniAdminBindIP(
|
||||
requestedBindIP,
|
||||
*miniAdminOptions.adminPassword != "",
|
||||
hasMTLS,
|
||||
allowInsecure,
|
||||
)
|
||||
miniAdminOptions.ip = &bindIP
|
||||
if bindIP != requestedBindIP {
|
||||
glog.Warningf(
|
||||
|
||||
@@ -17,17 +17,20 @@ var (
|
||||
)
|
||||
|
||||
type MessageQueueAgentOptions struct {
|
||||
brokers []pb.ServerAddress
|
||||
brokersString *string
|
||||
filerGroup *string
|
||||
ip *string
|
||||
port *int
|
||||
brokers []pb.ServerAddress
|
||||
brokersString *string
|
||||
filerGroup *string
|
||||
ip *string
|
||||
ipBind *string
|
||||
listenOnLocalhost *bool
|
||||
port *int
|
||||
}
|
||||
|
||||
func init() {
|
||||
cmdMqAgent.Run = runMqAgent // break init cycle
|
||||
mqAgentOptions.brokersString = cmdMqAgent.Flag.String("broker", "localhost:17777", "comma-separated message queue brokers")
|
||||
mqAgentOptions.ip = cmdMqAgent.Flag.String("ip", "", "message queue agent host address")
|
||||
mqAgentOptions.ipBind = cmdMqAgent.Flag.String("ip.bind", "", "ip address to bind to. If empty, default to same as -ip option.")
|
||||
mqAgentOptions.port = cmdMqAgent.Flag.Int("port", 16777, "message queue agent gRPC server port")
|
||||
}
|
||||
|
||||
@@ -60,8 +63,20 @@ func (mqAgentOpt *MessageQueueAgentOptions) startQueueAgent() bool {
|
||||
SeedBrokers: mqAgentOpt.brokers,
|
||||
}, grpcDialOption)
|
||||
|
||||
// The loopback companion listener is only kept while -ip.bind is unset and
|
||||
// the bind address falls back to -ip. An explicit -ip.bind pins the port to
|
||||
// exactly that address.
|
||||
bindIp := *mqAgentOpt.ip
|
||||
listenOnLocalhost := mqAgentOpt.ipBind == nil || *mqAgentOpt.ipBind == ""
|
||||
if mqAgentOpt.ipBind != nil && *mqAgentOpt.ipBind != "" {
|
||||
bindIp = *mqAgentOpt.ipBind
|
||||
}
|
||||
if mqAgentOpt.listenOnLocalhost != nil {
|
||||
listenOnLocalhost = *mqAgentOpt.listenOnLocalhost
|
||||
}
|
||||
|
||||
// start grpc listener
|
||||
grpcL, localL, err := util.NewIpAndLocalListeners(*mqAgentOpt.ip, *mqAgentOpt.port, 0)
|
||||
grpcL, localL, err := util.NewIpAndLocalListeners(bindIp, *mqAgentOpt.port, 0, listenOnLocalhost)
|
||||
if err != nil {
|
||||
glog.Fatalf("failed to listen on grpc port %d: %v", *mqAgentOpt.port, err)
|
||||
}
|
||||
@@ -84,7 +99,7 @@ func (mqAgentOpt *MessageQueueAgentOptions) startQueueAgent() bool {
|
||||
}()
|
||||
}
|
||||
|
||||
glog.Infof("Start Seaweed Message Queue Agent on %s:%d", *mqAgentOpt.ip, *mqAgentOpt.port)
|
||||
glog.Infof("Start Seaweed Message Queue Agent on %s:%d", bindIp, *mqAgentOpt.port)
|
||||
if err := grpcS.Serve(grpcL); err != nil && err != grpc.ErrServerStopped {
|
||||
glog.Errorf("MQ Agent failed to start: %v", err)
|
||||
}
|
||||
|
||||
+24
-14
@@ -18,18 +18,20 @@ var (
|
||||
)
|
||||
|
||||
type MessageQueueBrokerOptions struct {
|
||||
masters map[string]pb.ServerAddress
|
||||
mastersString *string
|
||||
filerGroup *string
|
||||
ip *string
|
||||
port *int
|
||||
dataCenter *string
|
||||
rack *string
|
||||
cpuprofile *string
|
||||
memprofile *string
|
||||
logFlushInterval *int
|
||||
debug *bool
|
||||
debugPort *int
|
||||
masters map[string]pb.ServerAddress
|
||||
mastersString *string
|
||||
filerGroup *string
|
||||
ip *string
|
||||
ipBind *string
|
||||
listenOnLocalhost *bool
|
||||
port *int
|
||||
dataCenter *string
|
||||
rack *string
|
||||
cpuprofile *string
|
||||
memprofile *string
|
||||
logFlushInterval *int
|
||||
debug *bool
|
||||
debugPort *int
|
||||
}
|
||||
|
||||
func init() {
|
||||
@@ -37,6 +39,7 @@ func init() {
|
||||
mqBrokerStandaloneOptions.mastersString = cmdMqBroker.Flag.String("master", "localhost:9333", "comma-separated master servers")
|
||||
mqBrokerStandaloneOptions.filerGroup = cmdMqBroker.Flag.String("filerGroup", "", "share metadata with other filers in the same filerGroup")
|
||||
mqBrokerStandaloneOptions.ip = cmdMqBroker.Flag.String("ip", util.DetectedHostAddress(), "broker host address")
|
||||
mqBrokerStandaloneOptions.ipBind = cmdMqBroker.Flag.String("ip.bind", "", "ip address to bind to. If empty, listen on all interfaces.")
|
||||
mqBrokerStandaloneOptions.port = cmdMqBroker.Flag.Int("port", 17777, "broker gRPC listen port")
|
||||
mqBrokerStandaloneOptions.dataCenter = cmdMqBroker.Flag.String("dataCenter", "", "prefer to read and write to volumes in this data center")
|
||||
mqBrokerStandaloneOptions.rack = cmdMqBroker.Flag.String("rack", "", "prefer to write to volumes in this rack")
|
||||
@@ -94,8 +97,15 @@ func (mqBrokerOpt *MessageQueueBrokerOptions) startQueueServer() bool {
|
||||
glog.Fatalf("failed to create new message broker for queue server: %v", err)
|
||||
}
|
||||
|
||||
// The loopback companion listener is only kept while -ip.bind is unset.
|
||||
// An explicit -ip.bind pins the port to exactly that address.
|
||||
listenOnLocalhost := *mqBrokerOpt.ipBind == ""
|
||||
if mqBrokerOpt.listenOnLocalhost != nil {
|
||||
listenOnLocalhost = *mqBrokerOpt.listenOnLocalhost
|
||||
}
|
||||
|
||||
// start grpc listener
|
||||
grpcL, localL, err := util.NewIpAndLocalListeners("", *mqBrokerOpt.port, 0)
|
||||
grpcL, localL, err := util.NewIpAndLocalListeners(*mqBrokerOpt.ipBind, *mqBrokerOpt.port, 0, listenOnLocalhost)
|
||||
if err != nil {
|
||||
glog.Fatalf("failed to listen on grpc port %d: %v", *mqBrokerOpt.port, err)
|
||||
}
|
||||
@@ -118,7 +128,7 @@ func (mqBrokerOpt *MessageQueueBrokerOptions) startQueueServer() bool {
|
||||
}()
|
||||
}
|
||||
|
||||
glog.V(0).Infof("MQ Broker listening on %s:%d", *mqBrokerOpt.ip, *mqBrokerOpt.port)
|
||||
glog.V(0).Infof("MQ Broker listening on %s:%d", *mqBrokerOpt.ipBind, *mqBrokerOpt.port)
|
||||
if err := grpcS.Serve(grpcL); err != nil && err != grpc.ErrServerStopped {
|
||||
glog.Errorf("Failed to serve MQ Broker: %v", err)
|
||||
}
|
||||
|
||||
+16
-9
@@ -46,6 +46,7 @@ type S3Options struct {
|
||||
filer *string
|
||||
ip *string
|
||||
bindIp *string
|
||||
listenOnLocalhost *bool
|
||||
port *int
|
||||
portHttps *int
|
||||
portGrpc *int
|
||||
@@ -359,6 +360,12 @@ func (s3opt *S3Options) startS3Server() bool {
|
||||
if *s3opt.portGrpc == 0 {
|
||||
*s3opt.portGrpc = 10000 + *s3opt.port
|
||||
}
|
||||
// The loopback companion listener is only kept while -ip.bind is unset.
|
||||
// An explicit -ip.bind pins each port to exactly that address.
|
||||
listenOnLocalhost := *s3opt.bindIp == ""
|
||||
if s3opt.listenOnLocalhost != nil {
|
||||
listenOnLocalhost = *s3opt.listenOnLocalhost
|
||||
}
|
||||
if *s3opt.bindIp == "" {
|
||||
*s3opt.bindIp = "0.0.0.0"
|
||||
}
|
||||
@@ -414,14 +421,14 @@ func (s3opt *S3Options) startS3Server() bool {
|
||||
|
||||
// Start Iceberg REST Catalog server if enabled
|
||||
if *s3opt.portIceberg > 0 {
|
||||
if httpS := s3opt.startIcebergServer(s3ApiServer); httpS != nil {
|
||||
if httpS := s3opt.startIcebergServer(s3ApiServer, listenOnLocalhost); httpS != nil {
|
||||
httpServers = append(httpServers, httpS)
|
||||
}
|
||||
}
|
||||
|
||||
// Start Lance Namespace server if enabled
|
||||
if s3opt.portLance != nil && *s3opt.portLance > 0 {
|
||||
if httpS := s3opt.startLanceServer(s3ApiServer); httpS != nil {
|
||||
if httpS := s3opt.startLanceServer(s3ApiServer, listenOnLocalhost); httpS != nil {
|
||||
httpServers = append(httpServers, httpS)
|
||||
}
|
||||
}
|
||||
@@ -445,7 +452,7 @@ func (s3opt *S3Options) startS3Server() bool {
|
||||
|
||||
listenAddress := fmt.Sprintf("%s:%d", *s3opt.bindIp, *s3opt.port)
|
||||
s3ApiListener, s3ApiLocalListener, err := util.NewIpAndLocalListeners(
|
||||
*s3opt.bindIp, *s3opt.port, time.Duration(*s3opt.idleTimeout)*time.Second)
|
||||
*s3opt.bindIp, *s3opt.port, time.Duration(*s3opt.idleTimeout)*time.Second, listenOnLocalhost)
|
||||
if err != nil {
|
||||
glog.Fatalf("S3 API Server listener on %s error: %v", listenAddress, err)
|
||||
}
|
||||
@@ -459,7 +466,7 @@ func (s3opt *S3Options) startS3Server() bool {
|
||||
|
||||
// starting grpc server
|
||||
grpcPort := *s3opt.portGrpc
|
||||
grpcL, grpcLocalL, err := util.NewIpAndLocalListeners(*s3opt.bindIp, grpcPort, 0)
|
||||
grpcL, grpcLocalL, err := util.NewIpAndLocalListeners(*s3opt.bindIp, grpcPort, 0, listenOnLocalhost)
|
||||
if err != nil {
|
||||
glog.Fatalf("s3 failed to listen on grpc port %d: %v", grpcPort, err)
|
||||
}
|
||||
@@ -532,7 +539,7 @@ func (s3opt *S3Options) startS3Server() bool {
|
||||
} else {
|
||||
glog.V(0).Infof("Start Seaweed S3 API Server %s at https port %d", version.Version(), *s3opt.portHttps)
|
||||
s3ApiListenerHttps, s3ApiLocalListenerHttps, err := util.NewIpAndLocalListeners(
|
||||
*s3opt.bindIp, *s3opt.portHttps, time.Duration(*s3opt.idleTimeout)*time.Second)
|
||||
*s3opt.bindIp, *s3opt.portHttps, time.Duration(*s3opt.idleTimeout)*time.Second, listenOnLocalhost)
|
||||
if err != nil {
|
||||
glog.Fatalf("S3 API HTTPS listener on %s:%d error: %v", *s3opt.bindIp, *s3opt.portHttps, err)
|
||||
}
|
||||
@@ -592,7 +599,7 @@ func (s3opt *S3Options) newShutdown(stopGrpc func(), s3ApiServer *s3api.S3ApiSer
|
||||
}
|
||||
|
||||
// startIcebergServer starts the Iceberg REST Catalog server on a separate port.
|
||||
func (s3opt *S3Options) startIcebergServer(s3ApiServer *s3api.S3ApiServer) *http.Server {
|
||||
func (s3opt *S3Options) startIcebergServer(s3ApiServer *s3api.S3ApiServer, listenOnLocalhost bool) *http.Server {
|
||||
icebergRouter := mux.NewRouter().SkipClean(true)
|
||||
// warehouse/parent query values may legally contain ';', which Go's
|
||||
// url.ParseQuery would otherwise drop
|
||||
@@ -614,7 +621,7 @@ func (s3opt *S3Options) startIcebergServer(s3ApiServer *s3api.S3ApiServer) *http
|
||||
|
||||
listenAddress := fmt.Sprintf("%s:%d", *s3opt.bindIp, *s3opt.portIceberg)
|
||||
icebergListener, icebergLocalListener, err := util.NewIpAndLocalListeners(
|
||||
*s3opt.bindIp, *s3opt.portIceberg, time.Duration(*s3opt.idleTimeout)*time.Second)
|
||||
*s3opt.bindIp, *s3opt.portIceberg, time.Duration(*s3opt.idleTimeout)*time.Second, listenOnLocalhost)
|
||||
if err != nil {
|
||||
// The catalog is auxiliary to the S3 gateway; a port conflict must not
|
||||
// take the whole server down.
|
||||
@@ -645,7 +652,7 @@ func (s3opt *S3Options) startIcebergServer(s3ApiServer *s3api.S3ApiServer) *http
|
||||
// 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) *http.Server {
|
||||
func (s3opt *S3Options) startLanceServer(s3ApiServer *s3api.S3ApiServer, listenOnLocalhost bool) *http.Server {
|
||||
lanceRouter := mux.NewRouter().SkipClean(true)
|
||||
lanceRouter.Use(util_http.EscapeSemicolonsInQuery)
|
||||
|
||||
@@ -660,7 +667,7 @@ func (s3opt *S3Options) startLanceServer(s3ApiServer *s3api.S3ApiServer) *http.S
|
||||
|
||||
listenAddress := fmt.Sprintf("%s:%d", *s3opt.bindIp, *s3opt.portLance)
|
||||
lanceListener, lanceLocalListener, err := util.NewIpAndLocalListeners(
|
||||
*s3opt.bindIp, *s3opt.portLance, time.Duration(*s3opt.idleTimeout)*time.Second)
|
||||
*s3opt.bindIp, *s3opt.portLance, time.Duration(*s3opt.idleTimeout)*time.Second, listenOnLocalhost)
|
||||
if err != nil {
|
||||
glog.Errorf("Lance Namespace listener on %s error: %v", listenAddress, err)
|
||||
return nil
|
||||
|
||||
@@ -42,14 +42,14 @@ func TestAuxListenerBindFailureIsNonFatal(t *testing.T) {
|
||||
}
|
||||
s3a := &s3api.S3ApiServer{}
|
||||
|
||||
if s := s3opt.startIcebergServer(s3a); s != nil {
|
||||
if s := s3opt.startIcebergServer(s3a, false); 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 {
|
||||
if s := s3opt.startLanceServer(s3a, false); s != nil {
|
||||
s.Close()
|
||||
t.Error("startLanceServer on an occupied port should return nil")
|
||||
}
|
||||
|
||||
+38
-1
@@ -12,6 +12,7 @@ import (
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb"
|
||||
stats_collect "github.com/seaweedfs/seaweedfs/weed/stats"
|
||||
"github.com/seaweedfs/seaweedfs/weed/util"
|
||||
flag "github.com/seaweedfs/seaweedfs/weed/util/fla9"
|
||||
"github.com/seaweedfs/seaweedfs/weed/util/grace"
|
||||
)
|
||||
|
||||
@@ -240,6 +241,17 @@ func init() {
|
||||
|
||||
}
|
||||
|
||||
// isServerFlagPassed checks if a specific flag was passed on the command line
|
||||
func isServerFlagPassed(name string) bool {
|
||||
found := false
|
||||
cmdServer.Flag.Visit(func(f *flag.Flag) {
|
||||
if f.Name == name {
|
||||
found = true
|
||||
}
|
||||
})
|
||||
return found
|
||||
}
|
||||
|
||||
func runServer(cmd *Command, args []string) bool {
|
||||
|
||||
if *serverOptions.debug {
|
||||
@@ -293,6 +305,11 @@ func runServer(cmd *Command, args []string) bool {
|
||||
actualPeersForComponents = *masterOptions.peers
|
||||
}
|
||||
|
||||
// The loopback companion listener is only kept while -ip.bind is unset and
|
||||
// the bind address falls back to -ip. Capture that here: serverBindIp is
|
||||
// repointed at serverIp below, after which the flag state is no longer
|
||||
// distinguishable.
|
||||
ipBindUnspecified := *serverBindIp == ""
|
||||
if *serverBindIp == "" {
|
||||
serverBindIp = serverIp
|
||||
}
|
||||
@@ -307,31 +324,51 @@ func runServer(cmd *Command, args []string) bool {
|
||||
// ip address
|
||||
masterOptions.ip = serverIp
|
||||
masterOptions.ipBind = serverBindIp
|
||||
masterOptions.listenOnLocalhost = &ipBindUnspecified
|
||||
// Use actualPeersForComponents for volume/filer, not masterOptions.peers which might be "none"
|
||||
filerOptions.masters = pb.ServerAddresses(actualPeersForComponents).ToServiceDiscovery()
|
||||
filerOptions.ip = serverIp
|
||||
filerOptions.bindIp = serverBindIp
|
||||
filerOptions.listenOnLocalhost = &ipBindUnspecified
|
||||
s3Options.ip = serverIp
|
||||
if *s3Options.bindIp == "" {
|
||||
s3Options.listenOnLocalhost = &ipBindUnspecified
|
||||
s3Options.bindIp = serverBindIp
|
||||
}
|
||||
if sftpOptions.bindIp == nil || *sftpOptions.bindIp == "" {
|
||||
sftpOptions.listenOnLocalhost = &ipBindUnspecified
|
||||
sftpOptions.bindIp = serverBindIp
|
||||
}
|
||||
iamOptions.ip = serverBindIp
|
||||
iamOptions.listenOnLocalhost = &ipBindUnspecified
|
||||
iamOptions.masters = &actualPeersForComponents
|
||||
webdavOptions.ipBind = serverBindIp
|
||||
serverOptions.v.ip = serverIp
|
||||
serverOptions.v.bindIp = serverBindIp
|
||||
serverOptions.v.listenOnLocalhost = &ipBindUnspecified
|
||||
serverOptions.v.masters = pb.ServerAddresses(actualPeersForComponents).ToAddresses()
|
||||
serverOptions.v.idleConnectionTimeout = serverTimeout
|
||||
serverOptions.v.dataCenter = serverDataCenter
|
||||
serverOptions.v.rack = serverRack
|
||||
mqBrokerOptions.ip = serverIp
|
||||
mqBrokerOptions.ipBind = serverBindIp
|
||||
mqBrokerOptions.listenOnLocalhost = &ipBindUnspecified
|
||||
mqBrokerOptions.masters = filerOptions.masters.GetInstancesAsMap()
|
||||
mqBrokerOptions.filerGroup = filerOptions.filerGroup
|
||||
mqAgentServerOptions.ip = serverIp
|
||||
mqAgentServerOptions.brokers = pb.ServerAddresses(*mqAgentServerOptions.brokersString).ToAddresses()
|
||||
mqAgentServerOptions.ipBind = serverBindIp
|
||||
mqAgentServerOptions.listenOnLocalhost = &ipBindUnspecified
|
||||
if isServerFlagPassed("mq.agent.brokers") {
|
||||
mqAgentServerOptions.brokers = pb.ServerAddresses(*mqAgentServerOptions.brokersString).ToAddresses()
|
||||
} else {
|
||||
// Default to the embedded broker's bind address: with an explicit
|
||||
// -ip.bind the broker does not listen on localhost.
|
||||
brokerHost := *serverBindIp
|
||||
if brokerHost == "0.0.0.0" || brokerHost == "::" || brokerHost == "[::]" {
|
||||
brokerHost = "localhost"
|
||||
}
|
||||
mqAgentServerOptions.brokers = []pb.ServerAddress{pb.NewServerAddress(brokerHost, *mqBrokerOptions.port, 0)}
|
||||
}
|
||||
|
||||
// serverOptions.v.pulseSeconds = pulseSeconds
|
||||
// masterOptions.pulseSeconds = pulseSeconds
|
||||
|
||||
@@ -25,6 +25,7 @@ var (
|
||||
type SftpOptions struct {
|
||||
filer *string
|
||||
bindIp *string
|
||||
listenOnLocalhost *bool
|
||||
port *int
|
||||
sshPrivateKey *string
|
||||
hostKeysFolder *string
|
||||
@@ -105,6 +106,12 @@ func (sftpOpt *SftpOptions) resolvePaths() {
|
||||
}
|
||||
|
||||
func (sftpOpt *SftpOptions) startSftpServer() bool {
|
||||
// The loopback companion listener is only kept while -ip.bind is unset.
|
||||
// An explicit -ip.bind pins the port to exactly that address.
|
||||
listenOnLocalhost := *sftpOpt.bindIp == ""
|
||||
if sftpOpt.listenOnLocalhost != nil {
|
||||
listenOnLocalhost = *sftpOpt.listenOnLocalhost
|
||||
}
|
||||
if *sftpOpt.bindIp == "" {
|
||||
*sftpOpt.bindIp = "0.0.0.0"
|
||||
}
|
||||
@@ -194,7 +201,7 @@ func (sftpOpt *SftpOptions) startSftpServer() bool {
|
||||
|
||||
// Start the SFTP service on TCP
|
||||
listenAddress := fmt.Sprintf("%s:%d", *sftpOpt.bindIp, *sftpOpt.port)
|
||||
sftpListener, sftpLocalListener, err := util.NewIpAndLocalListeners(*sftpOpt.bindIp, *sftpOpt.port, time.Duration(10)*time.Second)
|
||||
sftpListener, sftpLocalListener, err := util.NewIpAndLocalListeners(*sftpOpt.bindIp, *sftpOpt.port, time.Duration(10)*time.Second, listenOnLocalhost)
|
||||
if err != nil {
|
||||
glog.Fatalf("SFTP server listener on %s error: %v", listenAddress, err)
|
||||
}
|
||||
|
||||
+17
-9
@@ -45,6 +45,7 @@ type VolumeServerOptions struct {
|
||||
id *string
|
||||
publicUrl *string
|
||||
bindIp *string
|
||||
listenOnLocalhost *bool
|
||||
mastersString *string
|
||||
mserverString *string // deprecated, for backward compatibility
|
||||
masters []pb.ServerAddress
|
||||
@@ -397,6 +398,13 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v
|
||||
*v.ip = util.DetectedHostAddress()
|
||||
glog.V(0).Infof("detected volume server ip address: %v", *v.ip)
|
||||
}
|
||||
// The loopback companion listener is only kept while -ip.bind is unset
|
||||
// and the bind address falls back to -ip. An explicit -ip.bind pins each
|
||||
// port to exactly that address.
|
||||
listenOnLocalhost := *v.bindIp == ""
|
||||
if v.listenOnLocalhost != nil {
|
||||
listenOnLocalhost = *v.listenOnLocalhost
|
||||
}
|
||||
if *v.bindIp == "" {
|
||||
*v.bindIp = *v.ip
|
||||
}
|
||||
@@ -485,19 +493,19 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v
|
||||
v.syncBatchOption(),
|
||||
)
|
||||
// starting grpc server
|
||||
grpcS := v.startGrpcService(volumeServer)
|
||||
grpcS := v.startGrpcService(volumeServer, listenOnLocalhost)
|
||||
|
||||
// starting public http server
|
||||
var publicHttpDown httpdown.Server
|
||||
if v.isSeparatedPublicPort() {
|
||||
publicHttpDown = v.startPublicHttpService(publicVolumeMux)
|
||||
publicHttpDown = v.startPublicHttpService(publicVolumeMux, listenOnLocalhost)
|
||||
if nil == publicHttpDown {
|
||||
glog.Fatalf("start public http service failed")
|
||||
}
|
||||
}
|
||||
|
||||
// starting the cluster http server
|
||||
clusterHttpServer, closeCert := v.startClusterHttpService(volumeMux)
|
||||
clusterHttpServer, closeCert := v.startClusterHttpService(volumeMux, listenOnLocalhost)
|
||||
|
||||
grace.OnReload(volumeServer.LoadNewVolumes)
|
||||
grace.OnReload(volumeServer.Reload)
|
||||
@@ -632,9 +640,9 @@ func (m multiHttpdownServer) Stop() error {
|
||||
return firstErr
|
||||
}
|
||||
|
||||
func (v VolumeServerOptions) startGrpcService(vs volume_server_pb.VolumeServerServer) *grpc.Server {
|
||||
func (v VolumeServerOptions) startGrpcService(vs volume_server_pb.VolumeServerServer, listenOnLocalhost bool) *grpc.Server {
|
||||
grpcPort := *v.portGrpc
|
||||
grpcL, grpcLocalL, err := util.NewIpAndLocalListeners(*v.bindIp, grpcPort, 0)
|
||||
grpcL, grpcLocalL, err := util.NewIpAndLocalListeners(*v.bindIp, grpcPort, 0, listenOnLocalhost)
|
||||
if err != nil {
|
||||
glog.Fatalf("failed to listen on grpc port %d: %v", grpcPort, err)
|
||||
}
|
||||
@@ -653,10 +661,10 @@ func (v VolumeServerOptions) startGrpcService(vs volume_server_pb.VolumeServerSe
|
||||
return grpcS
|
||||
}
|
||||
|
||||
func (v VolumeServerOptions) startPublicHttpService(handler http.Handler) httpdown.Server {
|
||||
func (v VolumeServerOptions) startPublicHttpService(handler http.Handler, listenOnLocalhost bool) httpdown.Server {
|
||||
publicListeningAddress := util.JoinHostPort(*v.bindIp, *v.publicPort)
|
||||
glog.V(0).Infoln("Start Seaweed volume server", version.Version(), "public at", publicListeningAddress)
|
||||
publicListener, publicLocalListener, e := util.NewIpAndLocalListeners(*v.bindIp, *v.publicPort, time.Duration(*v.idleConnectionTimeout)*time.Second)
|
||||
publicListener, publicLocalListener, e := util.NewIpAndLocalListeners(*v.bindIp, *v.publicPort, time.Duration(*v.idleConnectionTimeout)*time.Second, listenOnLocalhost)
|
||||
if e != nil {
|
||||
glog.Fatalf("Volume server listener error:%v", e)
|
||||
}
|
||||
@@ -681,7 +689,7 @@ func (v VolumeServerOptions) startPublicHttpService(handler http.Handler) httpdo
|
||||
// for invoking the close func on every shutdown path — both the
|
||||
// SIGTERM/grace.OnInterrupt path and the shutdownCtx path used by
|
||||
// mini/integration tests.
|
||||
func (v VolumeServerOptions) startClusterHttpService(handler http.Handler) (httpdown.Server, func()) {
|
||||
func (v VolumeServerOptions) startClusterHttpService(handler http.Handler, listenOnLocalhost bool) (httpdown.Server, func()) {
|
||||
var (
|
||||
certFile, keyFile string
|
||||
)
|
||||
@@ -692,7 +700,7 @@ func (v VolumeServerOptions) startClusterHttpService(handler http.Handler) (http
|
||||
|
||||
listeningAddress := util.JoinHostPort(*v.bindIp, *v.port)
|
||||
glog.V(0).Infof("Start Seaweed volume server %s at %s", version.Version(), listeningAddress)
|
||||
listener, localListener, e := util.NewIpAndLocalListeners(*v.bindIp, *v.port, time.Duration(*v.idleConnectionTimeout)*time.Second)
|
||||
listener, localListener, e := util.NewIpAndLocalListeners(*v.bindIp, *v.port, time.Duration(*v.idleConnectionTimeout)*time.Second, listenOnLocalhost)
|
||||
if e != nil {
|
||||
glog.Fatalf("Volume server listener error:%v", e)
|
||||
}
|
||||
|
||||
@@ -154,7 +154,23 @@ func NewListener(addr string, timeout time.Duration) (ipListener net.Listener, e
|
||||
return
|
||||
}
|
||||
|
||||
func NewIpAndLocalListeners(host string, port int, timeout time.Duration) (ipListener net.Listener, localListener net.Listener, err error) {
|
||||
// IsLoopbackOrWildcardAddress reports whether host already covers loopback
|
||||
// traffic — empty (all interfaces), a wildcard address, or a loopback address —
|
||||
// in which case a separate localhost listener is unnecessary.
|
||||
func IsLoopbackOrWildcardAddress(host string) bool {
|
||||
switch host {
|
||||
case "", "localhost", "0.0.0.0", "127.0.0.1", "[::]", "[::1]":
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// NewIpAndLocalListeners binds host:port and, unless listenOnLocalhost is given
|
||||
// as false, a second listener on localhost:port so that local clients still
|
||||
// reach the service when it binds a specific address. Callers pass false when
|
||||
// the user explicitly chose the bind address (e.g. -ip.bind), honoring exactly
|
||||
// that address.
|
||||
func NewIpAndLocalListeners(host string, port int, timeout time.Duration, listenOnLocalhost ...bool) (ipListener net.Listener, localListener net.Listener, err error) {
|
||||
listener, err := net.Listen("tcp", JoinHostPort(host, port))
|
||||
if err != nil {
|
||||
return
|
||||
@@ -165,10 +181,10 @@ func NewIpAndLocalListeners(host string, port int, timeout time.Duration) (ipLis
|
||||
Timeout: timeout,
|
||||
}
|
||||
|
||||
if host != "localhost" && host != "" && host != "0.0.0.0" && host != "127.0.0.1" && host != "[::]" && host != "[::1]" {
|
||||
if (len(listenOnLocalhost) == 0 || listenOnLocalhost[0]) && !IsLoopbackOrWildcardAddress(host) {
|
||||
listener, err = net.Listen("tcp", JoinHostPort("localhost", port))
|
||||
if err != nil {
|
||||
glog.V(0).Infof("skip starting on %s:%d: %v", host, port, err)
|
||||
glog.V(0).Infof("skip starting localhost listener on :%d: %v", port, err)
|
||||
return ipListener, nil, nil
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user