From db5a1882b50561bf0d6df52c5dcd687b8bd3dc31 Mon Sep 17 00:00:00 2001 From: "Dr. Stefan Schimanski" Date: Wed, 16 Jan 2019 15:58:53 +0100 Subject: [PATCH] apiserver: sync with https server shutdown to flush existing connections Kubernetes-commit: 5b47f99164d76b6b043d7de2d3834abeda6685b0 --- pkg/server/deprecated_insecure_serving.go | 4 +++- pkg/server/genericapiserver.go | 9 ++++++-- pkg/server/genericapiserver_test.go | 5 +++-- pkg/server/secure_serving.go | 26 +++++++++++++---------- 4 files changed, 28 insertions(+), 16 deletions(-) diff --git a/pkg/server/deprecated_insecure_serving.go b/pkg/server/deprecated_insecure_serving.go index a78250eda..4f8d07a42 100644 --- a/pkg/server/deprecated_insecure_serving.go +++ b/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) { diff --git a/pkg/server/genericapiserver.go b/pkg/server/genericapiserver.go index 9dc4a3236..da3c6c045 100644 --- a/pkg/server/genericapiserver.go +++ b/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) }() diff --git a/pkg/server/genericapiserver_test.go b/pkg/server/genericapiserver_test.go index af1f93bb0..df6b0203c 100644 --- a/pkg/server/genericapiserver_test.go +++ b/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 { diff --git a/pkg/server/secure_serving.go b/pkg/server/secure_serving.go index 663c33bc4..b3dce77fd 100644 --- a/pkg/server/secure_serving.go +++ b/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 {