wait until all the on going requests go to a stable stat before shuti…

…ng down daemon

Signed-off-by: shenling.yyb <shenling.yyb@alibaba-inc.com>

.gitignore

@@ -14,3 +14,4 @@ coverage.txt
 bin/
 coverage/
 test/testdata/cp/*
+.vscode

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 {

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
 }

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.