volume: also listen on localhost like other components (#11995)

* volume: also listen on localhost like other components

Master, filer and s3 bind both the detected host ip and localhost via
NewIpAndLocalListeners, but the volume server used a plain NewListener on
bindIp, so a volume bound to the host ip never answers on loopback. Any
tooling that polls localhost (e.g. the java test workflow's readiness
curl on localhost:8080/status) then waits out its full timeout even
though the volume is up and registered with the master.

Use NewIpAndLocalListeners for the volume grpc, public and cluster http
listeners, and add a small multiHttpdownServer composite so graceful
shutdown still covers both listeners.

* volume: stop and wait dual listeners concurrently

A serial multiHttpdownServer.Stop let the localhost listener keep
accepting requests while the bound-ip listener drained, and a serial
Wait hid a localhost serving failure until the bound-ip listener ended.
Run both waits and both stops in parallel: Wait reports the first error
promptly, Stop closes every listener when shutdown begins.

weed/command/volume.go

@@ -573,15 +573,54 @@ func (v VolumeServerOptions) isSeparatedPublicPort() bool {
 	return *v.publicPort != *v.port
 }
 
+// multiHttpdownServer gracefully stops an http server serving on both the
+// bound ip and the localhost listeners as if it were a single server.
+type multiHttpdownServer []httpdown.Server
+
+// Wait returns the first member's non-nil error as soon as it happens, so a
+// failing localhost listener is reported even while the bound-ip listener
+// keeps serving, and nil once every member has finished cleanly.
+func (m multiHttpdownServer) Wait() error {
+	errs := make(chan error, len(m))
+	for _, s := range m {
+		go func(s httpdown.Server) { errs <- s.Wait() }(s)
+	}
+	for range m {
+		if err := <-errs; err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// Stop stops all members concurrently so the localhost listener closes as
+// soon as shutdown begins instead of after the bound-ip listener drains.
+func (m multiHttpdownServer) Stop() error {
+	errs := make(chan error, len(m))
+	for _, s := range m {
+		go func(s httpdown.Server) { errs <- s.Stop() }(s)
+	}
+	var firstErr error
+	for range m {
+		if err := <-errs; err != nil && firstErr == nil {
+			firstErr = err
+		}
+	}
+	return firstErr
+}
+
 func (v VolumeServerOptions) startGrpcService(vs volume_server_pb.VolumeServerServer) *grpc.Server {
 	grpcPort := *v.portGrpc
-	grpcL, err := util.NewListener(util.JoinHostPort(*v.bindIp, grpcPort), 0)
+	grpcL, grpcLocalL, err := util.NewIpAndLocalListeners(*v.bindIp, grpcPort, 0)
 	if err != nil {
 		glog.Fatalf("failed to listen on grpc port %d: %v", grpcPort, err)
 	}
 	grpcS := pb.NewGrpcServer(security.LoadServerTLS(util.GetViper(), "grpc.volume"))
 	volume_server_pb.RegisterVolumeServerServer(grpcS, vs)
 	reflection.Register(grpcS)
+	if grpcLocalL != nil {
+		go grpcS.Serve(grpcLocalL)
+	}
 	go func() {
 		if err := grpcS.Serve(grpcL); err != nil {
 			glog.Fatalf("start gRPC service failed, %s", err)
@@ -594,13 +633,16 @@ func (v VolumeServerOptions) startGrpcService(vs volume_server_pb.VolumeServerSe
 func (v VolumeServerOptions) startPublicHttpService(handler http.Handler) httpdown.Server {
 	publicListeningAddress := util.JoinHostPort(*v.bindIp, *v.publicPort)
 	glog.V(0).Infoln("Start Seaweed volume server", version.Version(), "public at", publicListeningAddress)
-	publicListener, e := util.NewListener(publicListeningAddress, time.Duration(*v.idleConnectionTimeout)*time.Second)
+	publicListener, publicLocalListener, e := util.NewIpAndLocalListeners(*v.bindIp, *v.publicPort, time.Duration(*v.idleConnectionTimeout)*time.Second)
 	if e != nil {
 		glog.Fatalf("Volume server listener error:%v", e)
 	}
 
 	pubHttp := httpdown.HTTP{StopTimeout: 5 * time.Minute, KillTimeout: 5 * time.Minute}
-	publicHttpDown := pubHttp.Serve(&http.Server{Handler: handler}, publicListener)
+	publicHttpDown := multiHttpdownServer{pubHttp.Serve(&http.Server{Handler: handler}, publicListener)}
+	if publicLocalListener != nil {
+		publicHttpDown = append(publicHttpDown, pubHttp.Serve(&http.Server{Handler: handler}, publicLocalListener))
+	}
 	go func() {
 		if err := publicHttpDown.Wait(); err != nil {
 			glog.Errorf("public http down wait failed, %v", err)
@@ -627,7 +669,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, e := util.NewListener(listeningAddress, time.Duration(*v.idleConnectionTimeout)*time.Second)
+	listener, localListener, e := util.NewIpAndLocalListeners(*v.bindIp, *v.port, time.Duration(*v.idleConnectionTimeout)*time.Second)
 	if e != nil {
 		glog.Fatalf("Volume server listener error:%v", e)
 	}
@@ -659,11 +701,15 @@ func (v VolumeServerOptions) startClusterHttpService(handler http.Handler) (http
 		httpS.TLSConfig.GetCertificate = getCert
 	}
 
-	clusterHttpServer := httpDown.Serve(httpS, listener)
+	clusterHttpServers := multiHttpdownServer{httpDown.Serve(httpS, listener)}
+	if localListener != nil {
+		localHttpS := &http.Server{Handler: handler, TLSConfig: httpS.TLSConfig}
+		clusterHttpServers = append(clusterHttpServers, httpDown.Serve(localHttpS, localListener))
+	}
 	go func() {
-		if e := clusterHttpServer.Wait(); e != nil {
+		if e := clusterHttpServers.Wait(); e != nil {
 			glog.Fatalf("Volume server fail to serve: %v", e)
 		}
 	}()
-	return clusterHttpServer, closeCert
+	return clusterHttpServers, closeCert
 }