Merge pull request #72970 from sttts/sttts-wait-for-http-server-finished-shutdown

apiserver: sync with http server shutdown to flush existing connections

cmd/cloud-controller-manager/app/controllermanager.go

@@ -145,7 +145,8 @@ func Run(c *cloudcontrollerconfig.CompletedConfig, stopCh <-chan struct{}) error
 	if c.SecureServing != nil {
 		unsecuredMux := genericcontrollermanager.NewBaseHandler(&c.ComponentConfig.Generic.Debugging, checks...)
 		handler := genericcontrollermanager.BuildHandlerChain(unsecuredMux, &c.Authorization, &c.Authentication)
-		if err := c.SecureServing.Serve(handler, 0, stopCh); err != nil {
+		// TODO: handle stoppedCh returned by c.SecureServing.Serve
+		if _, err := c.SecureServing.Serve(handler, 0, stopCh); err != nil {
 			return err
 		}
 	}

cmd/kube-controller-manager/app/controllermanager.go

@@ -170,7 +170,8 @@ func Run(c *config.CompletedConfig, stopCh <-chan struct{}) error {
 	if c.SecureServing != nil {
 		unsecuredMux = genericcontrollermanager.NewBaseHandler(&c.ComponentConfig.Generic.Debugging, checks...)
 		handler := genericcontrollermanager.BuildHandlerChain(unsecuredMux, &c.Authorization, &c.Authentication)
-		if err := c.SecureServing.Serve(handler, 0, stopCh); err != nil {
+		// TODO: handle stoppedCh returned by c.SecureServing.Serve
+		if _, err := c.SecureServing.Serve(handler, 0, stopCh); err != nil {
 			return err
 		}
 	}

cmd/kube-scheduler/app/server.go

@@ -208,7 +208,8 @@ func Run(cc schedulerserverconfig.CompletedConfig, stopCh <-chan struct{}) error
 	}
 	if cc.SecureServing != nil {
 		handler := buildHandlerChain(newHealthzHandler(&cc.ComponentConfig, false, checks...), cc.Authentication.Authenticator, cc.Authorization.Authorizer)
-		if err := cc.SecureServing.Serve(handler, 0, stopCh); err != nil {
+		// TODO: handle stoppedCh returned by c.SecureServing.Serve
+		if _, err := cc.SecureServing.Serve(handler, 0, stopCh); err != nil {
 			// fail early for secure handlers, removing the old error loop from above
 			return fmt.Errorf("failed to start healthz server: %v", err)
 		}

staging/src/k8s.io/apiserver/pkg/server/deprecated_insecure_serving.go

@@ -50,7 +50,9 @@ func (s *DeprecatedInsecureServingInfo) Serve(handler http.Handler, shutdownTime
 	} else {
 		klog.Infof("Serving insecurely on %s", s.Listener.Addr())
 	}
-	return RunServer(insecureServer, s.Listener, shutdownTimeout, stopCh)
+	_, err := RunServer(insecureServer, s.Listener, shutdownTimeout, stopCh)
+	// NOTE: we do not handle stoppedCh returned by RunServer for graceful termination here
+	return err
 }
 
 func (s *DeprecatedInsecureServingInfo) NewLoopbackClientConfig() (*rest.Config, error) {

staging/src/k8s.io/apiserver/pkg/server/genericapiserver.go

@@ -291,9 +291,11 @@ func (s preparedGenericAPIServer) NonBlockingRun(stopCh <-chan struct{}) error {
 
 	// Use an internal stop channel to allow cleanup of the listeners on error.
 	internalStopCh := make(chan struct{})
-
+	var stoppedCh <-chan struct{}
 	if s.SecureServingInfo != nil && s.Handler != nil {
-		if err := s.SecureServingInfo.Serve(s.Handler, s.ShutdownTimeout, internalStopCh); err != nil {
+		var err error
+		stoppedCh, err = s.SecureServingInfo.Serve(s.Handler, s.ShutdownTimeout, internalStopCh)
+		if err != nil {
 			close(internalStopCh)
 			return err
 		}
@@ -305,6 +307,9 @@ func (s preparedGenericAPIServer) NonBlockingRun(stopCh <-chan struct{}) error {
 	go func() {
 		<-stopCh
 		close(internalStopCh)
+		if stoppedCh != nil {
+			<-stoppedCh
+		}
 		s.HandlerChainWaitGroup.Wait()
 		close(auditStopCh)
 	}()

staging/src/k8s.io/apiserver/pkg/server/genericapiserver_test.go

@@ -539,9 +539,9 @@ func TestGracefulShutdown(t *testing.T) {
 
 	// get port
 	serverPort := ln.Addr().(*net.TCPAddr).Port
-	err = RunServer(insecureServer, ln, 10*time.Second, stopCh)
+	stoppedCh, err := RunServer(insecureServer, ln, 10*time.Second, stopCh)
 	if err != nil {
-		t.Errorf("RunServer err: %v", err)
+		t.Fatalf("RunServer err: %v", err)
 	}
 
 	graceCh := make(chan struct{})
@@ -568,6 +568,7 @@ func TestGracefulShutdown(t *testing.T) {
 
 	// wait for wait group handler finish
 	s.HandlerChainWaitGroup.Wait()
+	<-stoppedCh
 
 	// check server all handlers finished.
 	if !graceShutdown {

staging/src/k8s.io/apiserver/pkg/server/secure_serving.go

@@ -37,12 +37,12 @@ const (
 	defaultKeepAlivePeriod = 3 * time.Minute
 )
 
-// serveSecurely runs the secure http server. It fails only if certificates cannot
-// be loaded or the initial listen call fails. The actual server loop (stoppable by closing
-// stopCh) runs in a go routine, i.e. serveSecurely does not block.
-func (s *SecureServingInfo) Serve(handler http.Handler, shutdownTimeout time.Duration, stopCh <-chan struct{}) error {
+// Serve runs the secure http server. It fails only if certificates cannot be loaded or the initial listen call fails.
+// The actual server loop (stoppable by closing stopCh) runs in a go routine, i.e. Serve does not block.
+// It returns a stoppedCh that is closed when all non-hijacked active requests have been processed.
+func (s *SecureServingInfo) Serve(handler http.Handler, shutdownTimeout time.Duration, stopCh <-chan struct{}) (<-chan struct{}, error) {
 	if s.Listener == nil {
-		return fmt.Errorf("listener must not be nil")
+		return nil, fmt.Errorf("listener must not be nil")
 	}
 
 	secureServer := &http.Server{
@@ -110,7 +110,7 @@ func (s *SecureServingInfo) Serve(handler http.Handler, shutdownTimeout time.Dur
 
 	// apply settings to the server
 	if err := http2.ConfigureServer(secureServer, http2Options); err != nil {
-		return fmt.Errorf("error configuring http2: %v", err)
+		return nil, fmt.Errorf("error configuring http2: %v", err)
 	}
 
 	klog.Infof("Serving securely on %s", secureServer.Addr)
@@ -118,21 +118,25 @@ func (s *SecureServingInfo) Serve(handler http.Handler, shutdownTimeout time.Dur
 }
 
 // RunServer listens on the given port if listener is not given,
-// then spawns a go-routine continuously serving
-// until the stopCh is closed. This function does not block.
+// then spawns a go-routine continuously serving until the stopCh is closed.
+// It returns a stoppedCh that is closed when all non-hijacked active requests
+// have been processed.
+// This function does not block
 // TODO: make private when insecure serving is gone from the kube-apiserver
 func RunServer(
 	server *http.Server,
 	ln net.Listener,
 	shutDownTimeout time.Duration,
 	stopCh <-chan struct{},
-) error {
+) (<-chan struct{}, error) {
 	if ln == nil {
-		return fmt.Errorf("listener must not be nil")
+		return nil, fmt.Errorf("listener must not be nil")
 	}
 
 	// Shutdown server gracefully.
+	stoppedCh := make(chan struct{})
 	go func() {
+		defer close(stoppedCh)
 		<-stopCh
 		ctx, cancel := context.WithTimeout(context.Background(), shutDownTimeout)
 		server.Shutdown(ctx)
@@ -159,7 +163,7 @@ func RunServer(
 		}
 	}()
 
-	return nil
+	return stoppedCh, nil
 }
 
 type NamedTLSCert struct {