diff --git a/.gitignore b/.gitignore index 8ba661616..90ec6d871 100644 --- a/.gitignore +++ b/.gitignore @@ -14,3 +14,4 @@ coverage.txt bin/ coverage/ test/testdata/cp/* +.vscode diff --git a/apis/server/router.go b/apis/server/router.go index 1bffc53cf..521442054 100644 --- a/apis/server/router.go +++ b/apis/server/router.go @@ -6,6 +6,7 @@ import ( "fmt" "net/http" "net/http/pprof" + "sync/atomic" "time" serverTypes "github.com/alibaba/pouch/apis/server/types" @@ -177,6 +178,9 @@ func filter(handler serverTypes.Handler, s *Server) http.HandlerFunc { ctx, cancel := context.WithCancel(pctx) defer cancel() + atomic.AddInt32(&s.FlyingReq, 1) + defer atomic.AddInt32(&s.FlyingReq, -1) + s.lock.RLock() if len(s.ManagerWhiteList) > 0 && req.TLS != nil && len(req.TLS.PeerCertificates) > 0 { if _, isManager := s.ManagerWhiteList[req.TLS.PeerCertificates[0].Subject.CommonName]; !isManager { diff --git a/apis/server/server.go b/apis/server/server.go index eb3c70d4a..9bfb20216 100644 --- a/apis/server/server.go +++ b/apis/server/server.go @@ -7,6 +7,7 @@ import ( "net/http" "strings" "sync" + "sync/atomic" "time" "github.com/alibaba/pouch/cri/stream" @@ -33,6 +34,7 @@ type Server struct { APIPlugin hookplugins.APIPlugin ManagerWhiteList map[string]struct{} lock sync.RWMutex + FlyingReq int32 } // Start setup route table and listen to specified address which currently only supports unix socket and tcp address. @@ -107,5 +109,24 @@ func (s *Server) Stop() error { for _, one := range s.listeners { one.Close() } + + // drain all requests on going or timeout after one minute + drain := make(chan struct{}) + go func() { + for { + if atomic.LoadInt32(&s.FlyingReq) == 0 { + close(drain) + return + } + time.Sleep(time.Microsecond * 50) + } + }() + + select { + case <-drain: + case <-time.After(60 * time.Second): + logrus.Errorf("stop pouch server after waited 60 seconds, on going request %d", atomic.LoadInt32(&s.FlyingReq)) + } + return nil } diff --git a/main.go b/main.go index 08f04c40a..67960ace2 100644 --- a/main.go +++ b/main.go @@ -247,7 +247,7 @@ func runDaemon(cmd *cobra.Command) error { } signal.Notify(signalCh, syscall.SIGINT, syscall.SIGQUIT, syscall.SIGTERM, syscall.SIGHUP) - sigHandles = append(sigHandles, d.ShutdownPlugin, d.Shutdown) + sigHandles = append(sigHandles, d.Shutdown, d.ShutdownPlugin) go func() { // FIXME: I think the Run() should always return error.