Merge pull request kubernetes#125483 from wojtek-t/storage_readiness_hook

Implement resilient watchcache initialization post-start-hook

Author: k8s-ci-robot

cmd/kube-apiserver/app/options/options_test.go

@@ -139,16 +139,17 @@ func TestAddFlags(t *testing.T) {
 	expected := &ServerRunOptions{
 		Options: &controlplaneapiserver.Options{
 			GenericServerRunOptions: &apiserveroptions.ServerRunOptions{
-				AdvertiseAddress:            netutils.ParseIPSloppy("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:       int64(3 * 1024 * 1024),
-				MaxRequestBodyBytes:         int64(3 * 1024 * 1024),
-				ComponentGlobalsRegistry:    componentGlobalsRegistry,
-				ComponentName:               utilversion.DefaultKubeComponent,
+				AdvertiseAddress:             netutils.ParseIPSloppy("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,
+				StorageInitializationTimeout: time.Minute,
+				JSONPatchMaxCopyBytes:        int64(3 * 1024 * 1024),
+				MaxRequestBodyBytes:          int64(3 * 1024 * 1024),
+				ComponentGlobalsRegistry:     componentGlobalsRegistry,
+				ComponentName:                utilversion.DefaultKubeComponent,
 			},
 			Admission: &kubeoptions.AdmissionOptions{
 				GenericAdmission: &apiserveroptions.AdmissionOptions{

pkg/controlplane/apiserver/options/options_test.go

@@ -123,16 +123,17 @@ func TestAddFlags(t *testing.T) {
 	// This is a snapshot of expected options parsed by args.
 	expected := &Options{
 		GenericServerRunOptions: &apiserveroptions.ServerRunOptions{
-			AdvertiseAddress:            netutils.ParseIPSloppy("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:       int64(3 * 1024 * 1024),
-			MaxRequestBodyBytes:         int64(3 * 1024 * 1024),
-			ComponentGlobalsRegistry:    componentGlobalsRegistry,
-			ComponentName:               utilversion.DefaultKubeComponent,
+			AdvertiseAddress:             netutils.ParseIPSloppy("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,
+			StorageInitializationTimeout: time.Minute,
+			JSONPatchMaxCopyBytes:        int64(3 * 1024 * 1024),
+			MaxRequestBodyBytes:          int64(3 * 1024 * 1024),
+			ComponentGlobalsRegistry:     componentGlobalsRegistry,
+			ComponentName:                utilversion.DefaultKubeComponent,
 		},
 		Admission: &kubeoptions.AdmissionOptions{
 			GenericAdmission: &apiserveroptions.AdmissionOptions{

pkg/controlplane/apiserver/server.go

@@ -245,6 +245,10 @@ func (c completedConfig) New(name string, delegationTarget genericapiserver.Dele
 		})
 	}
 
+	if utilfeature.DefaultFeatureGate.Enabled(apiserverfeatures.WatchCacheInitializationPostStartHook) {
+		s.GenericAPIServer.AddPostStartHookOrDie("storage-readiness", s.GenericAPIServer.StorageReadinessHook.Hook)
+	}
+
 	s.GenericAPIServer.AddPostStartHookOrDie("start-legacy-token-tracking-controller", func(hookContext genericapiserver.PostStartHookContext) error {
 		go legacytokentracking.NewController(client).Run(hookContext.StopCh)
 		return nil

pkg/features/kube_features.go

@@ -1275,6 +1275,8 @@ var defaultKubernetesFeatureGates = map[featuregate.Feature]featuregate.FeatureS
 
 	genericfeatures.WatchBookmark: {Default: true, PreRelease: featuregate.GA, LockToDefault: true},
 
+	genericfeatures.WatchCacheInitializationPostStartHook: {Default: false, PreRelease: featuregate.Beta},
+
 	genericfeatures.WatchFromStorageWithoutResourceVersion: {Default: false, PreRelease: featuregate.Beta},
 
 	genericfeatures.WatchList: {Default: true, PreRelease: featuregate.Beta},

staging/src/k8s.io/apiserver/pkg/features/kube_features.go

@@ -290,6 +290,12 @@ const (
 	// Enables support for watch bookmark events.
 	WatchBookmark featuregate.Feature = "WatchBookmark"
 
+	// owner: @wojtek-t
+	// beta: v1.31
+	//
+	// Enables post-start-hook for storage readiness
+	WatchCacheInitializationPostStartHook featuregate.Feature = "WatchCacheInitializationPostStartHook"
+
 	// owner: @serathius
 	// beta: 1.30
 	// Enables watches without resourceVersion to be served from storage.
@@ -408,6 +414,8 @@ var defaultKubernetesFeatureGates = map[featuregate.Feature]featuregate.FeatureS
 
 	WatchBookmark: {Default: true, PreRelease: featuregate.GA, LockToDefault: true},
 
+	WatchCacheInitializationPostStartHook: {Default: false, PreRelease: featuregate.Beta},
+
 	WatchFromStorageWithoutResourceVersion: {Default: false, PreRelease: featuregate.Beta},
 
 	InPlacePodVerticalScaling: {Default: false, PreRelease: featuregate.Alpha},

staging/src/k8s.io/apiserver/pkg/registry/generic/registry/store.go

@@ -226,6 +226,10 @@ type Store struct {
 	// storageVersionHash as empty in the discovery document.
 	StorageVersioner runtime.GroupVersioner
 
+	// ReadinessCheckFunc checks if the storage is ready for accepting requests.
+	// The field is optional, if set needs to be thread-safe.
+	ReadinessCheckFunc func() error
+
 	// DestroyFunc cleans up clients used by the underlying Storage; optional.
 	// If set, DestroyFunc has to be implemented in thread-safe way and
 	// be prepared for being called more than once.
@@ -234,6 +238,7 @@ type Store struct {
 
 // Note: the rest.StandardStorage interface aggregates the common REST verbs
 var _ rest.StandardStorage = &Store{}
+var _ rest.StorageWithReadiness = &Store{}
 var _ rest.TableConvertor = &Store{}
 var _ GenericStore = &Store{}
 
@@ -292,6 +297,14 @@ func (e *Store) New() runtime.Object {
 	return e.NewFunc()
 }
 
+// ReadinessCheck checks if the storage is ready for accepting requests.
+func (e *Store) ReadinessCheck() error {
+	if e.ReadinessCheckFunc != nil {
+		return e.ReadinessCheckFunc()
+	}
+	return nil
+}
+
 // Destroy cleans up its resources on shutdown.
 func (e *Store) Destroy() {
 	if e.DestroyFunc != nil {
@@ -1614,6 +1627,9 @@ func (e *Store) CompleteWithOptions(options *generic.StoreOptions) error {
 			}
 		}
 	}
+	if e.Storage.Storage != nil {
+		e.ReadinessCheckFunc = e.Storage.Storage.ReadinessCheck
+	}
 
 	return nil
 }

staging/src/k8s.io/apiserver/pkg/registry/rest/rest.go

@@ -52,6 +52,8 @@ import (
 // Storage is a generic interface for RESTful storage services.
 // Resources which are exported to the RESTful API of apiserver need to implement this interface. It is expected
 // that objects may implement any of the below interfaces.
+//
+// Consider using StorageWithReadiness whenever possible.
 type Storage interface {
 	// New returns an empty object that can be used with Create and Update after request data has been put into it.
 	// This object must be a pointer type for use with Codec.DecodeInto([]byte, runtime.Object)
@@ -63,6 +65,14 @@ type Storage interface {
 	Destroy()
 }
 
+// StorageWithReadiness extends Storage interface with the readiness check.
+type StorageWithReadiness interface {
+	Storage
+
+	// ReadinessCheck allows for checking storage readiness.
+	ReadinessCheck() error
+}
+
 // Scoper indicates what scope the resource is at. It must be specified.
 // It is usually provided automatically based on your strategy.
 type Scoper interface {

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

@@ -216,6 +216,10 @@ type Config struct {
 	// twice this value.  Note that it is up to the request handlers to ignore or honor this timeout. In seconds.
 	MinRequestTimeout int
 
+	// StorageInitializationTimeout defines the maximum amount of time to wait for storage initialization
+	// before declaring apiserver ready.
+	StorageInitializationTimeout time.Duration
+
 	// This represents the maximum amount of time it should take for apiserver to complete its startup
 	// sequence and become healthy. From apiserver's start time to when this amount of time has
 	// elapsed, /livez will assume that unfinished post-start hooks will complete successfully and
@@ -426,6 +430,7 @@ func NewConfig(codecs serializer.CodecFactory) *Config {
 		MaxMutatingRequestsInFlight:    200,
 		RequestTimeout:                 time.Duration(60) * time.Second,
 		MinRequestTimeout:              1800,
+		StorageInitializationTimeout:   time.Minute,
 		LivezGracePeriod:               time.Duration(0),
 		ShutdownDelayDuration:          time.Duration(0),
 		// 1.5MB is the default client request size in bytes
@@ -824,6 +829,7 @@ func (c completedConfig) New(name string, delegationTarget DelegationTarget) (*G
 		ShutdownSendRetryAfter: c.ShutdownSendRetryAfter,
 
 		APIServerID:           c.APIServerID,
+		StorageReadinessHook:  NewStorageReadinessHook(c.StorageInitializationTimeout),
 		StorageVersionManager: c.StorageVersionManager,
 
 		EffectiveVersion: c.EffectiveVersion,

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

@@ -233,6 +233,10 @@ type GenericAPIServer struct {
 	// APIServerID is the ID of this API server
 	APIServerID string
 
+	// StorageReadinessHook implements post-start-hook functionality for checking readiness
+	// of underlying storage for registered resources.
+	StorageReadinessHook *StorageReadinessHook
+
 	// StorageVersionManager holds the storage versions of the API resources installed by this server.
 	StorageVersionManager storageversion.Manager
 
@@ -844,6 +848,7 @@ func (s *GenericAPIServer) InstallLegacyAPIGroup(apiPrefix string, apiGroupInfo
 	} else {
 		s.Handler.GoRestfulContainer.Add(legacyRootAPIHandler.WebService())
 	}
+	s.registerStorageReadinessCheck("", apiGroupInfo)
 
 	return nil
 }
@@ -902,10 +907,28 @@ func (s *GenericAPIServer) InstallAPIGroups(apiGroupInfos ...*APIGroupInfo) erro
 
 		s.DiscoveryGroupManager.AddGroup(apiGroup)
 		s.Handler.GoRestfulContainer.Add(discovery.NewAPIGroupHandler(s.Serializer, apiGroup).WebService())
+		s.registerStorageReadinessCheck(apiGroupInfo.PrioritizedVersions[0].Group, apiGroupInfo)
 	}
 	return nil
 }
 
+// registerStorageReadinessCheck registers the readiness checks for all underlying storages
+// for a given APIGroup.
+func (s *GenericAPIServer) registerStorageReadinessCheck(groupName string, apiGroupInfo *APIGroupInfo) {
+	for version, storageMap := range apiGroupInfo.VersionedResourcesStorageMap {
+		for resource, storage := range storageMap {
+			if withReadiness, ok := storage.(rest.StorageWithReadiness); ok {
+				gvr := metav1.GroupVersionResource{
+					Group:    groupName,
+					Version:  version,
+					Resource: resource,
+				}
+				s.StorageReadinessHook.RegisterStorage(gvr, withReadiness)
+			}
+		}
+	}
+}
+
 // InstallAPIGroup exposes the given api group in the API.
 // The <apiGroupInfo> passed into this function shouldn't be used elsewhere as the
 // underlying storage will be destroyed on this servers shutdown.

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

@@ -47,16 +47,17 @@ const (
 type ServerRunOptions struct {
 	AdvertiseAddress net.IP
 
-	CorsAllowedOriginList       []string
-	HSTSDirectives              []string
-	ExternalHost                string
-	MaxRequestsInFlight         int
-	MaxMutatingRequestsInFlight int
-	RequestTimeout              time.Duration
-	GoawayChance                float64
-	LivezGracePeriod            time.Duration
-	MinRequestTimeout           int
-	ShutdownDelayDuration       time.Duration
+	CorsAllowedOriginList        []string
+	HSTSDirectives               []string
+	ExternalHost                 string
+	MaxRequestsInFlight          int
+	MaxMutatingRequestsInFlight  int
+	RequestTimeout               time.Duration
+	GoawayChance                 float64
+	LivezGracePeriod             time.Duration
+	MinRequestTimeout            int
+	StorageInitializationTimeout time.Duration
+	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
@@ -116,6 +117,7 @@ func NewServerRunOptionsForComponent(componentName string, componentGlobalsRegis
 		RequestTimeout:                      defaults.RequestTimeout,
 		LivezGracePeriod:                    defaults.LivezGracePeriod,
 		MinRequestTimeout:                   defaults.MinRequestTimeout,
+		StorageInitializationTimeout:        defaults.StorageInitializationTimeout,
 		ShutdownDelayDuration:               defaults.ShutdownDelayDuration,
 		ShutdownWatchTerminationGracePeriod: defaults.ShutdownWatchTerminationGracePeriod,
 		JSONPatchMaxCopyBytes:               defaults.JSONPatchMaxCopyBytes,
@@ -140,6 +142,7 @@ func (s *ServerRunOptions) ApplyTo(c *server.Config) error {
 	c.RequestTimeout = s.RequestTimeout
 	c.GoawayChance = s.GoawayChance
 	c.MinRequestTimeout = s.MinRequestTimeout
+	c.StorageInitializationTimeout = s.StorageInitializationTimeout
 	c.ShutdownDelayDuration = s.ShutdownDelayDuration
 	c.JSONPatchMaxCopyBytes = s.JSONPatchMaxCopyBytes
 	c.MaxRequestBodyBytes = s.MaxRequestBodyBytes
@@ -197,6 +200,10 @@ func (s *ServerRunOptions) Validate() []error {
 		errors = append(errors, fmt.Errorf("--min-request-timeout can not be negative value"))
 	}
 
+	if s.StorageInitializationTimeout < 0 {
+		errors = append(errors, fmt.Errorf("--storage-initialization-timeout can not be negative value"))
+	}
+
 	if s.ShutdownDelayDuration < 0 {
 		errors = append(errors, fmt.Errorf("--shutdown-delay-duration can not be negative value"))
 	}
@@ -350,6 +357,9 @@ func (s *ServerRunOptions) AddUniversalFlags(fs *pflag.FlagSet) {
 		"handler, which picks a randomized value above this number as the connection timeout, "+
 		"to spread out load.")
 
+	fs.DurationVar(&s.StorageInitializationTimeout, "storage-initialization-timeout", s.StorageInitializationTimeout,
+		"Maximum amount of time to wait for storage initialization before declaring apiserver ready. Defaults to 1m.")
+
 	fs.DurationVar(&s.ShutdownDelayDuration, "shutdown-delay-duration", s.ShutdownDelayDuration, ""+
 		"Time to delay the termination. During that time the server keeps serving requests normally. The endpoints /healthz and /livez "+
 		"will return success, but /readyz immediately returns failure. Graceful termination starts after this delay "+