MyFavMirrors/minio
1
0
Fork 0
mirror of https://github.com/minio/minio.git synced 2025-01-23 20:53:18 -05:00
Code Issues Packages Projects Releases Wiki Activity

minhttp: Remove probe usage, move to golang error. (#1459)

Browse Source 
Fixes #1454
This commit is contained in:
Harshavardhana 2016-05-03 01:07:34 -07:00
parent ad8e27a963
commit 7ae40eb1bb
4 changed files with 53 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
pkg/minhttp/http_nix.go
View File

@ -34,7 +34,6 @@ import (
"time" "time"
"github.com/facebookgo/httpdown" "github.com/facebookgo/httpdown"
"github.com/minio/minio/pkg/probe"
) )
// An app contains one or more servers and their associated configuration. // An app contains one or more servers and their associated configuration.
@ -43,15 +42,15 @@ type app struct {
listeners []net.Listener listeners []net.Listener
sds []httpdown.Server sds []httpdown.Server
net *minNet net *minNet
errors chan *probe.Error errors chan error
} }
// listen initailize listeners // listen initailize listeners
func (a *app) listen() *probe.Error { func (a *app) listen() error {
for _, s := range a.servers { for _, s := range a.servers {
l, err := a.net.Listen("tcp", s.Addr) l, err := a.net.Listen("tcp", s.Addr)
if err != nil { if err != nil {
return err.Trace() return err
} }
if s.TLSConfig != nil { if s.TLSConfig != nil {
l = tls.NewListener(l, s.TLSConfig) l = tls.NewListener(l, s.TLSConfig)
@ -81,7 +80,7 @@ func (a *app) wait() {
go func(s httpdown.Server) { go func(s httpdown.Server) {
defer wg.Done() defer wg.Done()
if err := s.Wait(); err != nil { if err := s.Wait(); err != nil {
a.errors <- probe.NewError(err) a.errors <- err
} }
}(s) }(s)
} }
@ -103,7 +102,7 @@ func (a *app) trapSignal(wg *sync.WaitGroup) {
go func(s httpdown.Server) { go func(s httpdown.Server) {
defer wg.Done() defer wg.Done()
if err := s.Stop(); err != nil { if err := s.Stop(); err != nil {
a.errors <- probe.NewError(err) a.errors <- err
} }
}(s) }(s)
} }
@ -112,7 +111,7 @@ func (a *app) trapSignal(wg *sync.WaitGroup) {
// we only return here if there's an error, otherwise the new process // we only return here if there's an error, otherwise the new process
// will send us a TERM when it's ready to trigger the actual shutdown. // will send us a TERM when it's ready to trigger the actual shutdown.
if _, err := a.net.StartProcess(); err != nil { if _, err := a.net.StartProcess(); err != nil {
a.errors <- err.Trace() a.errors <- err
} }
} }
} }
@ -120,7 +119,7 @@ func (a *app) trapSignal(wg *sync.WaitGroup) {
// ListenAndServe will serve the given http.Servers and will monitor for signals // ListenAndServe will serve the given http.Servers and will monitor for signals
// allowing for graceful termination (SIGTERM) or restart (SIGUSR2/SIGHUP). // allowing for graceful termination (SIGTERM) or restart (SIGUSR2/SIGHUP).
func ListenAndServe(servers ...*http.Server) *probe.Error { func ListenAndServe(servers ...*http.Server) error {
// get parent process id // get parent process id
ppid := os.Getppid() ppid := os.Getppid()
@ -129,12 +128,12 @@ func ListenAndServe(servers ...*http.Server) *probe.Error {
listeners: make([]net.Listener, 0, len(servers)), listeners: make([]net.Listener, 0, len(servers)),
sds: make([]httpdown.Server, 0, len(servers)), sds: make([]httpdown.Server, 0, len(servers)),
net: &minNet{}, net: &minNet{},
errors: make(chan *probe.Error, 1+(len(servers)*2)), errors: make(chan error, 1+(len(servers)*2)),
} }
// Acquire Listeners // Acquire Listeners
if err := a.listen(); err != nil { if err := a.listen(); err != nil {
return err.Trace() return err
} }
// Start serving. // Start serving.
@ -143,7 +142,7 @@ func ListenAndServe(servers ...*http.Server) *probe.Error {
// Close the parent if we inherited and it wasn't init that started us. // Close the parent if we inherited and it wasn't init that started us.
if os.Getenv("LISTEN_FDS") != "" && ppid != 1 { if os.Getenv("LISTEN_FDS") != "" && ppid != 1 {
if err := syscall.Kill(ppid, syscall.SIGTERM); err != nil { if err := syscall.Kill(ppid, syscall.SIGTERM); err != nil {
return probe.NewError(err) return err
} }
} }
@ -160,14 +159,14 @@ func ListenAndServe(servers ...*http.Server) *probe.Error {
if err == nil { if err == nil {
panic("unexpected nil error") panic("unexpected nil error")
} }
return err.Trace() return err
case <-waitdone: case <-waitdone:
return nil return nil
} }
} }
// ListenAndServeLimited is similar to ListenAndServe but ratelimited with connLimit value // ListenAndServeLimited is similar to ListenAndServe but ratelimited with connLimit value
func ListenAndServeLimited(connLimit int, servers ...*http.Server) *probe.Error { func ListenAndServeLimited(connLimit int, servers ...*http.Server) error {
// get parent process id // get parent process id
ppid := os.Getppid() ppid := os.Getppid()
@ -176,12 +175,12 @@ func ListenAndServeLimited(connLimit int, servers ...*http.Server) *probe.Error
listeners: make([]net.Listener, 0, len(servers)), listeners: make([]net.Listener, 0, len(servers)),
sds: make([]httpdown.Server, 0, len(servers)), sds: make([]httpdown.Server, 0, len(servers)),
net: &minNet{connLimit: connLimit}, net: &minNet{connLimit: connLimit},
errors: make(chan *probe.Error, 1+(len(servers)*2)), errors: make(chan error, 1+(len(servers)*2)),
} }
// Acquire Listeners // Acquire Listeners
if err := a.listen(); err != nil { if err := a.listen(); err != nil {
return err.Trace() return err
} }
// Start serving. // Start serving.
@ -190,7 +189,7 @@ func ListenAndServeLimited(connLimit int, servers ...*http.Server) *probe.Error
// Close the parent if we inherited and it wasn't init that started us. // Close the parent if we inherited and it wasn't init that started us.
if os.Getenv("LISTEN_FDS") != "" && ppid != 1 { if os.Getenv("LISTEN_FDS") != "" && ppid != 1 {
if err := syscall.Kill(ppid, syscall.SIGTERM); err != nil { if err := syscall.Kill(ppid, syscall.SIGTERM); err != nil {
return probe.NewError(err) return err
} }
} }
@ -207,7 +206,7 @@ func ListenAndServeLimited(connLimit int, servers ...*http.Server) *probe.Error
if err == nil { if err == nil {
panic("unexpected nil error") panic("unexpected nil error")
} }
return err.Trace() return err
case <-waitdone: case <-waitdone:
return nil return nil
} }

33
pkg/minhttp/http_windows.go
View File

@ -34,7 +34,6 @@ import (
"time" "time"
"github.com/facebookgo/httpdown" "github.com/facebookgo/httpdown"
"github.com/minio/minio/pkg/probe"
) )
// An app contains one or more servers and their associated configuration. // An app contains one or more servers and their associated configuration.
@ -43,15 +42,15 @@ type app struct {
listeners []net.Listener listeners []net.Listener
sds []httpdown.Server sds []httpdown.Server
net *minNet net *minNet
errors chan *probe.Error errors chan error
} }
// listen initailize listeners // listen initailize listeners
func (a *app) listen() *probe.Error { func (a *app) listen() error {
for _, s := range a.servers { for _, s := range a.servers {
l, err := a.net.Listen("tcp", s.Addr) l, err := a.net.Listen("tcp", s.Addr)
if err != nil { if err != nil {
return err.Trace() return err
} }
if s.TLSConfig != nil { if s.TLSConfig != nil {
l = tls.NewListener(l, s.TLSConfig) l = tls.NewListener(l, s.TLSConfig)
@ -81,7 +80,7 @@ func (a *app) wait() {
go func(s httpdown.Server) { go func(s httpdown.Server) {
defer wg.Done() defer wg.Done()
if err := s.Wait(); err != nil { if err := s.Wait(); err != nil {
a.errors <- probe.NewError(err) a.errors <- err
} }
}(s) }(s)
} }
@ -103,7 +102,7 @@ func (a *app) trapSignal(wg *sync.WaitGroup) {
go func(s httpdown.Server) { go func(s httpdown.Server) {
defer wg.Done() defer wg.Done()
if err := s.Stop(); err != nil { if err := s.Stop(); err != nil {
a.errors <- probe.NewError(err) a.errors <- err
} }
}(s) }(s)
} }
@ -112,7 +111,7 @@ func (a *app) trapSignal(wg *sync.WaitGroup) {
// we only return here if there's an error, otherwise the new process // we only return here if there's an error, otherwise the new process
// will send us a TERM when it's ready to trigger the actual shutdown. // will send us a TERM when it's ready to trigger the actual shutdown.
if _, err := a.net.StartProcess(); err != nil { if _, err := a.net.StartProcess(); err != nil {
a.errors <- err.Trace() a.errors <- err
} }
} }
} }
@ -120,7 +119,7 @@ func (a *app) trapSignal(wg *sync.WaitGroup) {
// ListenAndServe will serve the given http.Servers and will monitor for signals // ListenAndServe will serve the given http.Servers and will monitor for signals
// allowing for graceful termination (SIGTERM) or restart (SIGUSR2/SIGHUP). // allowing for graceful termination (SIGTERM) or restart (SIGUSR2/SIGHUP).
func ListenAndServe(servers ...*http.Server) *probe.Error { func ListenAndServe(servers ...*http.Server) error {
// get parent process id // get parent process id
ppid := os.Getppid() ppid := os.Getppid()
@ -129,12 +128,12 @@ func ListenAndServe(servers ...*http.Server) *probe.Error {
listeners: make([]net.Listener, 0, len(servers)), listeners: make([]net.Listener, 0, len(servers)),
sds: make([]httpdown.Server, 0, len(servers)), sds: make([]httpdown.Server, 0, len(servers)),
net: &minNet{}, net: &minNet{},
errors: make(chan *probe.Error, 1+(len(servers)*2)), errors: make(chan error, 1+(len(servers)*2)),
} }
// Acquire Listeners // Acquire Listeners
if err := a.listen(); err != nil { if err := a.listen(); err != nil {
return err.Trace() return err
} }
// Start serving. // Start serving.
@ -143,7 +142,7 @@ func ListenAndServe(servers ...*http.Server) *probe.Error {
// Close the parent if we inherited and it wasn't init that started us. // Close the parent if we inherited and it wasn't init that started us.
if os.Getenv("LISTEN_FDS") != "" && ppid != 1 { if os.Getenv("LISTEN_FDS") != "" && ppid != 1 {
if err := terminateProcess(ppid, 1); err != nil { if err := terminateProcess(ppid, 1); err != nil {
return probe.NewError(err) return err
} }
} }
@ -160,14 +159,14 @@ func ListenAndServe(servers ...*http.Server) *probe.Error {
if err == nil { if err == nil {
panic("unexpected nil error") panic("unexpected nil error")
} }
return err.Trace() return err
case <-waitdone: case <-waitdone:
return nil return nil
} }
} }
// ListenAndServeLimited is similar to ListenAndServe but ratelimited with connLimit value // ListenAndServeLimited is similar to ListenAndServe but ratelimited with connLimit value
func ListenAndServeLimited(connLimit int, servers ...*http.Server) *probe.Error { func ListenAndServeLimited(connLimit int, servers ...*http.Server) error {
// get parent process id // get parent process id
ppid := os.Getppid() ppid := os.Getppid()
@ -176,12 +175,12 @@ func ListenAndServeLimited(connLimit int, servers ...*http.Server) *probe.Error
listeners: make([]net.Listener, 0, len(servers)), listeners: make([]net.Listener, 0, len(servers)),
sds: make([]httpdown.Server, 0, len(servers)), sds: make([]httpdown.Server, 0, len(servers)),
net: &minNet{connLimit: connLimit}, net: &minNet{connLimit: connLimit},
errors: make(chan *probe.Error, 1+(len(servers)*2)), errors: make(chan error, 1+(len(servers)*2)),
} }
// Acquire Listeners // Acquire Listeners
if err := a.listen(); err != nil { if err := a.listen(); err != nil {
return err.Trace() return err
} }
// Start serving. // Start serving.
@ -190,7 +189,7 @@ func ListenAndServeLimited(connLimit int, servers ...*http.Server) *probe.Error
// Close the parent if we inherited and it wasn't init that started us. // Close the parent if we inherited and it wasn't init that started us.
if os.Getenv("LISTEN_FDS") != "" && ppid != 1 { if os.Getenv("LISTEN_FDS") != "" && ppid != 1 {
if err := terminateProcess(ppid, 1); err != nil { if err := terminateProcess(ppid, 1); err != nil {
return probe.NewError(err) return err
} }
} }
@ -207,7 +206,7 @@ func ListenAndServeLimited(connLimit int, servers ...*http.Server) *probe.Error
if err == nil { if err == nil {
panic("unexpected nil error") panic("unexpected nil error")
} }
return err.Trace() return err
case <-waitdone: case <-waitdone:
return nil return nil
} }

42
pkg/minhttp/net.go
View File

@ -24,8 +24,6 @@ import (
"strconv" "strconv"
"strings" "strings"
"sync" "sync"
"github.com/minio/minio/pkg/probe"
) )
// This package is a fork https://github.com/facebookgo/grace // This package is a fork https://github.com/facebookgo/grace
@ -71,8 +69,8 @@ type fileListener interface {
} }
// getInheritedListeners - look for LISTEN_FDS in environment variables and populate listeners accordingly // getInheritedListeners - look for LISTEN_FDS in environment variables and populate listeners accordingly
func (n *minNet) getInheritedListeners() *probe.Error { func (n *minNet) getInheritedListeners() error {
var retErr *probe.Error var retErr error
n.inheritOnce.Do(func() { n.inheritOnce.Do(func() {
n.mutex.Lock() n.mutex.Lock()
defer n.mutex.Unlock() defer n.mutex.Unlock()
@ -82,7 +80,7 @@ func (n *minNet) getInheritedListeners() *probe.Error {
} }
count, err := strconv.Atoi(countStr) count, err := strconv.Atoi(countStr)
if err != nil { if err != nil {
retErr = probe.NewError(fmt.Errorf("found invalid count value: %s=%s", envCountKey, countStr)) retErr = fmt.Errorf("found invalid count value: %s=%s", envCountKey, countStr)
return return
} }
@ -92,18 +90,18 @@ func (n *minNet) getInheritedListeners() *probe.Error {
l, err := net.FileListener(file) l, err := net.FileListener(file)
if err != nil { if err != nil {
file.Close() file.Close()
retErr = probe.NewError(err) retErr = err
return return
} }
if err := file.Close(); err != nil { if err := file.Close(); err != nil {
retErr = probe.NewError(err) retErr = err
return return
} }
n.inheritedListeners = append(n.inheritedListeners, l) n.inheritedListeners = append(n.inheritedListeners, l)
} }
}) })
if retErr != nil { if retErr != nil {
return retErr.Trace() return retErr
} }
return nil return nil
} }
@ -112,20 +110,20 @@ func (n *minNet) getInheritedListeners() *probe.Error {
// a stream-oriented network: "tcp", "tcp4", "tcp6", "unix" or "unixpacket". It // a stream-oriented network: "tcp", "tcp4", "tcp6", "unix" or "unixpacket". It
// returns an inherited net.Listener for the matching network and address, or // returns an inherited net.Listener for the matching network and address, or
// creates a new one using net.Listen() // creates a new one using net.Listen()
func (n *minNet) Listen(nett, laddr string) (net.Listener, *probe.Error) { func (n *minNet) Listen(nett, laddr string) (net.Listener, error) {
switch nett { switch nett {
default: default:
return nil, probe.NewError(net.UnknownNetworkError(nett)) return nil, net.UnknownNetworkError(nett)
case "tcp", "tcp4", "tcp6": case "tcp", "tcp4", "tcp6":
addr, err := net.ResolveTCPAddr(nett, laddr) addr, err := net.ResolveTCPAddr(nett, laddr)
if err != nil { if err != nil {
return nil, probe.NewError(err) return nil, err
} }
return n.ListenTCP(nett, addr) return n.ListenTCP(nett, addr)
case "unix", "unixpacket": case "unix", "unixpacket":
addr, err := net.ResolveUnixAddr(nett, laddr) addr, err := net.ResolveUnixAddr(nett, laddr)
if err != nil { if err != nil {
return nil, probe.NewError(err) return nil, err
} }
return n.ListenUnix(nett, addr) return n.ListenUnix(nett, addr)
} }
@ -134,9 +132,9 @@ func (n *minNet) Listen(nett, laddr string) (net.Listener, *probe.Error) {
// ListenTCP announces on the local network address laddr. The network net must // ListenTCP announces on the local network address laddr. The network net must
// be: "tcp", "tcp4" or "tcp6". It returns an inherited net.Listener for the // be: "tcp", "tcp4" or "tcp6". It returns an inherited net.Listener for the
// matching network and address, or creates a new one using net.ListenTCP. // matching network and address, or creates a new one using net.ListenTCP.
func (n *minNet) ListenTCP(nett string, laddr *net.TCPAddr) (net.Listener, *probe.Error) { func (n *minNet) ListenTCP(nett string, laddr *net.TCPAddr) (net.Listener, error) {
if err := n.getInheritedListeners(); err != nil { if err := n.getInheritedListeners(); err != nil {
return nil, err.Trace() return nil, err
} }
n.mutex.Lock() n.mutex.Lock()
@ -158,7 +156,7 @@ func (n *minNet) ListenTCP(nett string, laddr *net.TCPAddr) (net.Listener, *prob
// make a fresh listener // make a fresh listener
l, err := net.ListenTCP(nett, laddr) l, err := net.ListenTCP(nett, laddr)
if err != nil { if err != nil {
return nil, probe.NewError(err) return nil, err
} }
n.activeListeners = append(n.activeListeners, rateLimitedListener(l, n.connLimit)) n.activeListeners = append(n.activeListeners, rateLimitedListener(l, n.connLimit))
return l, nil return l, nil
@ -167,9 +165,9 @@ func (n *minNet) ListenTCP(nett string, laddr *net.TCPAddr) (net.Listener, *prob
// ListenUnix announces on the local network address laddr. The network net // ListenUnix announces on the local network address laddr. The network net
// must be a: "unix" or "unixpacket". It returns an inherited net.Listener for // must be a: "unix" or "unixpacket". It returns an inherited net.Listener for
// the matching network and address, or creates a new one using net.ListenUnix. // the matching network and address, or creates a new one using net.ListenUnix.
func (n *minNet) ListenUnix(nett string, laddr *net.UnixAddr) (net.Listener, *probe.Error) { func (n *minNet) ListenUnix(nett string, laddr *net.UnixAddr) (net.Listener, error) {
if err := n.getInheritedListeners(); err != nil { if err := n.getInheritedListeners(); err != nil {
return nil, err.Trace() return nil, err
} }
n.mutex.Lock() n.mutex.Lock()
@ -191,7 +189,7 @@ func (n *minNet) ListenUnix(nett string, laddr *net.UnixAddr) (net.Listener, *pr
// make a fresh listener // make a fresh listener
l, err := net.ListenUnix(nett, laddr) l, err := net.ListenUnix(nett, laddr)
if err != nil { if err != nil {
return nil, probe.NewError(err) return nil, err
} }
n.activeListeners = append(n.activeListeners, rateLimitedListener(l, n.connLimit)) n.activeListeners = append(n.activeListeners, rateLimitedListener(l, n.connLimit))
return l, nil return l, nil
@ -232,7 +230,7 @@ func (n1 minAddr) IsEqual(n2 net.Addr) bool {
// arguments as when it was originally started. This allows for a newly // arguments as when it was originally started. This allows for a newly
// deployed binary to be started. It returns the pid of the newly started // deployed binary to be started. It returns the pid of the newly started
// process when successful. // process when successful.
func (n *minNet) StartProcess() (int, *probe.Error) { func (n *minNet) StartProcess() (int, error) {
listeners := n.getActiveListeners() listeners := n.getActiveListeners()
// Extract the fds from the listeners. // Extract the fds from the listeners.
files := make([]*os.File, len(listeners)) files := make([]*os.File, len(listeners))
@ -240,7 +238,7 @@ func (n *minNet) StartProcess() (int, *probe.Error) {
var err error var err error
files[i], err = l.(fileListener).File() files[i], err = l.(fileListener).File()
if err != nil { if err != nil {
return 0, probe.NewError(err) return 0, err
} }
defer files[i].Close() defer files[i].Close()
} }
@ -249,7 +247,7 @@ func (n *minNet) StartProcess() (int, *probe.Error) {
// the file it points to has been changed we will use the updated symlink. // the file it points to has been changed we will use the updated symlink.
argv0, err := exec.LookPath(os.Args[0]) argv0, err := exec.LookPath(os.Args[0])
if err != nil { if err != nil {
return 0, probe.NewError(err) return 0, err
} }
// Pass on the environment and replace the old count key with the new one. // Pass on the environment and replace the old count key with the new one.
@ -268,7 +266,7 @@ func (n *minNet) StartProcess() (int, *probe.Error) {
Files: allFiles, Files: allFiles,
}) })
if err != nil { if err != nil {
return 0, probe.NewError(err) return 0, err
} }
return process.Pid, nil return process.Pid, nil
} }

2
server-main.go
View File

@ -306,5 +306,5 @@ func serverMain(c *cli.Context) {
// Start server. // Start server.
err := minhttp.ListenAndServe(apiServer) err := minhttp.ListenAndServe(apiServer)
errorIf(err.Cause, "Failed to start the minio server.", nil) errorIf(err, "Failed to start the minio server.", nil)
} }