apiserver: add --shutdown-delay-duration to keep serving until LBs stop serving traffic

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

@@ -165,6 +165,11 @@ type Config struct {
 	// elapsed, /healthz will assume that unfinished post-start hooks will complete successfully and
 	// therefore return true.
 	MaxStartupSequenceDuration time.Duration
+	// ShutdownDelayDuration allows to block shutdown for some time, e.g. until endpoints pointing to this API server
+	// have converged on all node. During this time, the API server keeps serving, /healthz will return 200,
+	// but /readyz will return failure.
+	ShutdownDelayDuration time.Duration
+
 	// The limit on the total size increase all "copy" operations in a json
 	// patch may cause.
 	// This affects all places that applies json patch in the binary.
@@ -283,6 +288,7 @@ func NewConfig(codecs serializer.CodecFactory) *Config {
 		RequestTimeout:              time.Duration(60) * time.Second,
 		MinRequestTimeout:           1800,
 		MaxStartupSequenceDuration:  time.Duration(0),
+		ShutdownDelayDuration:       time.Duration(0),
 		// 10MB is the recommended maximum client request size in bytes
 		// the etcd server should accept. See
 		// https://github.com/etcd-io/etcd/blob/release-3.3/etcdserver/server.go#L90.
@@ -489,10 +495,11 @@ func (c completedConfig) New(name string, delegationTarget DelegationTarget) (*G
 		EquivalentResourceRegistry: c.EquivalentResourceRegistry,
 		HandlerChainWaitGroup:      c.HandlerChainWaitGroup,
 
-		minRequestTimeout: time.Duration(c.MinRequestTimeout) * time.Second,
-		ShutdownTimeout:   c.RequestTimeout,
-		SecureServingInfo: c.SecureServing,
-		ExternalAddress:   c.ExternalAddress,
+		minRequestTimeout:     time.Duration(c.MinRequestTimeout) * time.Second,
+		ShutdownTimeout:       c.RequestTimeout,
+		ShutdownDelayDuration: c.ShutdownDelayDuration,
+		SecureServingInfo:     c.SecureServing,
+		ExternalAddress:       c.ExternalAddress,
 
 		Handler: apiServerHandler,
 

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

@@ -180,6 +180,11 @@ type GenericAPIServer struct {
 	// HandlerChainWaitGroup allows you to wait for all chain handlers finish after the server shutdown.
 	HandlerChainWaitGroup *utilwaitgroup.SafeWaitGroup
 
+	// ShutdownDelayDuration allows to block shutdown for some time, e.g. until endpoints pointing to this API server
+	// have converged on all node. During this time, the API server keeps serving, /healthz will return 200,
+	// but /readyz will return failure.
+	ShutdownDelayDuration time.Duration
+
 	// The limit on the request body size that would be accepted and decoded in a write request.
 	// 0 means no limit.
 	maxRequestBodyBytes int64
@@ -287,18 +292,32 @@ func (s *GenericAPIServer) PrepareRun() preparedGenericAPIServer {
 // Run spawns the secure http server. It only returns if stopCh is closed
 // or the secure port cannot be listened on initially.
 func (s preparedGenericAPIServer) Run(stopCh <-chan struct{}) error {
-	err := s.NonBlockingRun(stopCh)
+	delayedStopCh := make(chan struct{})
+
+	go func() {
+		defer close(delayedStopCh)
+		<-stopCh
+
+		time.Sleep(s.ShutdownDelayDuration)
+	}()
+
+	// close socket after delayed stopCh
+	err := s.NonBlockingRun(delayedStopCh)
 	if err != nil {
 		return err
 	}
 
 	<-stopCh
 
+	// run shutdown hooks directly. This includes deregistering from the kubernetes endpoint in case of kube-apiserver.
 	err = s.RunPreShutdownHooks()
 	if err != nil {
 		return err
 	}
 
+	// wait for the delayed stopCh before closing the handler chain (it rejects everything after Wait has been called).
+	<-delayedStopCh
+
 	// Wait for all requests to finish, which are bounded by the RequestTimeout variable.
 	s.HandlerChainWaitGroup.Wait()
 

staging/src/k8s.io/apiserver/pkg/server/options/server_run_options.go

@@ -43,6 +43,7 @@ type ServerRunOptions struct {
 	RequestTimeout              time.Duration
 	MaxStartupSequenceDuration  time.Duration
 	MinRequestTimeout           int
+	ShutdownDelayDuration       time.Duration
 	// We intentionally did not add a flag for this option. Users of the
 	// apiserver library can wire it to a flag.
 	JSONPatchMaxCopyBytes int64
@@ -63,6 +64,7 @@ func NewServerRunOptions() *ServerRunOptions {
 		RequestTimeout:              defaults.RequestTimeout,
 		MaxStartupSequenceDuration:  defaults.MaxStartupSequenceDuration,
 		MinRequestTimeout:           defaults.MinRequestTimeout,
+		ShutdownDelayDuration:       defaults.ShutdownDelayDuration,
 		JSONPatchMaxCopyBytes:       defaults.JSONPatchMaxCopyBytes,
 		MaxRequestBodyBytes:         defaults.MaxRequestBodyBytes,
 	}
@@ -77,6 +79,7 @@ func (s *ServerRunOptions) ApplyTo(c *server.Config) error {
 	c.MaxStartupSequenceDuration = s.MaxStartupSequenceDuration
 	c.RequestTimeout = s.RequestTimeout
 	c.MinRequestTimeout = s.MinRequestTimeout
+	c.ShutdownDelayDuration = s.ShutdownDelayDuration
 	c.JSONPatchMaxCopyBytes = s.JSONPatchMaxCopyBytes
 	c.MaxRequestBodyBytes = s.MaxRequestBodyBytes
 	c.PublicAddress = s.AdvertiseAddress
@@ -143,6 +146,10 @@ func (s *ServerRunOptions) Validate() []error {
 		errors = append(errors, fmt.Errorf("--min-request-timeout can not be negative value"))
 	}
 
+	if s.ShutdownDelayDuration < 0 {
+		errors = append(errors, fmt.Errorf("--shutdown-delay-duration can not be negative value"))
+	}
+
 	if s.JSONPatchMaxCopyBytes < 0 {
 		errors = append(errors, fmt.Errorf("--json-patch-max-copy-bytes can not be negative value"))
 	}
@@ -206,5 +213,10 @@ func (s *ServerRunOptions) AddUniversalFlags(fs *pflag.FlagSet) {
 	fs.BoolVar(&s.EnableInfightQuotaHandler, "enable-inflight-quota-handler", s.EnableInfightQuotaHandler, ""+
 		"If true, replace the max-in-flight handler with an enhanced one that queues and dispatches with priority and fairness")
 
+	fs.DurationVar(&s.ShutdownDelayDuration, "shutdown-delay-duration", s.ShutdownDelayDuration, ""+
+		"Time to delay the termination. During that time the server keeps serving requests normally and /healthz "+
+		"returns success, but /readzy immediately returns failure. Graceful termination starts after this delay "+
+		"has elapsed. This can be used to allow load balancer to stop sending traffic to this server.")
+
 	utilfeature.DefaultMutableFeatureGate.AddFlag(fs)
 }

staging/src/k8s.io/apiserver/pkg/server/options/server_run_options_test.go

@@ -152,6 +152,22 @@ func TestServerRunOptionsValidate(t *testing.T) {
 			},
 			expectErr: "--maximum-startup-sequence-duration can not be a negative value",
 		},
+		{
+			name: "Test when MinimalShutdownDuration is negative value",
+			testOptions: &ServerRunOptions{
+				AdvertiseAddress:            net.ParseIP("192.168.10.10"),
+				CorsAllowedOriginList:       []string{"10.10.10.100", "10.10.10.200"},
+				MaxRequestsInFlight:         400,
+				MaxMutatingRequestsInFlight: 200,
+				RequestTimeout:              time.Duration(2) * time.Minute,
+				MinRequestTimeout:           1800,
+				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
+				MaxRequestBodyBytes:         10 * 1024 * 1024,
+				TargetRAMMB:                 65536,
+				ShutdownDelayDuration:       -time.Second,
+			},
+			expectErr: "--shutdown-delay-duration can not be negative value",
+		},
 		{
 			name: "Test when ServerRunOptions is valid",
 			testOptions: &ServerRunOptions{