Merge pull request kubernetes#129205 from tosi3k/wc-configurable-retention

Configure watch cache history window based on request timeout

Author: k8s-ci-robot

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

@@ -175,6 +175,7 @@ func TestAddFlags(t *testing.T) {
 					CompactionInterval:    storagebackend.DefaultCompactInterval,
 					CountMetricPollPeriod: time.Minute,
 					DBMetricPollInterval:  storagebackend.DefaultDBMetricPollInterval,
+					EventsHistoryWindow:   storagebackend.DefaultEventsHistoryWindow,
 					HealthcheckTimeout:    storagebackend.DefaultHealthcheckTimeout,
 					ReadycheckTimeout:     storagebackend.DefaultReadinessTimeout,
 					LeaseManagerConfig: etcd3.LeaseManagerConfig{

pkg/controlplane/apiserver/options/options.go

@@ -225,6 +225,16 @@ func (o *Options) Complete(ctx context.Context, fss cliflag.NamedFlagSets, alter
 		return CompletedOptions{}, fmt.Errorf("error creating self-signed certificates: %v", err)
 	}
 
+	if o.GenericServerRunOptions.RequestTimeout > 0 {
+		// Setting the EventsHistoryWindow as a maximum of the value set in the
+		// watchcache-specific options and the value of the request timeout plus
+		// some epsilon.
+		// This is done to make sure that the list+watch pattern can still be
+		// usable in large clusters with the elevated request timeout where the
+		// initial list can take a considerable amount of time.
+		completed.Etcd.StorageConfig.EventsHistoryWindow = max(completed.Etcd.StorageConfig.EventsHistoryWindow, completed.GenericServerRunOptions.RequestTimeout+15*time.Second)
+	}
+
 	if len(completed.GenericServerRunOptions.ExternalHost) == 0 {
 		if len(completed.GenericServerRunOptions.AdvertiseAddress) > 0 {
 			completed.GenericServerRunOptions.ExternalHost = completed.GenericServerRunOptions.AdvertiseAddress.String()

pkg/controlplane/apiserver/options/options_test.go

@@ -167,6 +167,7 @@ func TestAddFlags(t *testing.T) {
 				CompactionInterval:    storagebackend.DefaultCompactInterval,
 				CountMetricPollPeriod: time.Minute,
 				DBMetricPollInterval:  storagebackend.DefaultDBMetricPollInterval,
+				EventsHistoryWindow:   storagebackend.DefaultEventsHistoryWindow,
 				HealthcheckTimeout:    storagebackend.DefaultHealthcheckTimeout,
 				ReadycheckTimeout:     storagebackend.DefaultReadinessTimeout,
 				LeaseManagerConfig: etcd3.LeaseManagerConfig{

staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/customresource_handler_test.go

@@ -58,6 +58,7 @@ import (
 	genericregistry "k8s.io/apiserver/pkg/registry/generic/registry"
 	"k8s.io/apiserver/pkg/registry/rest"
 	"k8s.io/apiserver/pkg/server/options"
+	"k8s.io/apiserver/pkg/storage/cacher"
 	etcd3testing "k8s.io/apiserver/pkg/storage/etcd3/testing"
 	"k8s.io/apiserver/pkg/util/webhook"
 	"k8s.io/client-go/tools/cache"
@@ -484,6 +485,10 @@ func testHandlerConversion(t *testing.T, enableWatchCache bool) {
 		t.Fatal(err)
 	}
 
+	if enableWatchCache {
+		storageConfig.EventsHistoryWindow = cacher.DefaultEventFreshDuration
+	}
+
 	etcdOptions := options.NewEtcdOptions(storageConfig)
 	etcdOptions.StorageConfig.Codec = unstructured.UnstructuredJSONScheme
 	restOptionsGetter := generic.RESTOptions{

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

@@ -54,17 +54,18 @@ func StorageWithCacher() generic.StorageDecorator {
 		}
 
 		cacherConfig := cacherstorage.Config{
-			Storage:        s,
-			Versioner:      storage.APIObjectVersioner{},
-			GroupResource:  storageConfig.GroupResource,
-			ResourcePrefix: resourcePrefix,
-			KeyFunc:        keyFunc,
-			NewFunc:        newFunc,
-			NewListFunc:    newListFunc,
-			GetAttrsFunc:   getAttrsFunc,
-			IndexerFuncs:   triggerFuncs,
-			Indexers:       indexers,
-			Codec:          storageConfig.Codec,
+			Storage:             s,
+			Versioner:           storage.APIObjectVersioner{},
+			GroupResource:       storageConfig.GroupResource,
+			EventsHistoryWindow: storageConfig.EventsHistoryWindow,
+			ResourcePrefix:      resourcePrefix,
+			KeyFunc:             keyFunc,
+			NewFunc:             newFunc,
+			NewListFunc:         newListFunc,
+			GetAttrsFunc:        getAttrsFunc,
+			IndexerFuncs:        triggerFuncs,
+			Indexers:            indexers,
+			Codec:               storageConfig.Codec,
 		}
 		cacher, err := cacherstorage.NewCacherFromConfig(cacherConfig)
 		if err != nil {

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

@@ -2435,15 +2435,16 @@ func newTestGenericStoreRegistry(t *testing.T, scheme *runtime.Scheme, hasCacheE
 	}
 	if hasCacheEnabled {
 		config := cacherstorage.Config{
-			Storage:        s,
-			Versioner:      storage.APIObjectVersioner{},
-			GroupResource:  schema.GroupResource{Resource: "pods"},
-			ResourcePrefix: podPrefix,
-			KeyFunc:        func(obj runtime.Object) (string, error) { return storage.NoNamespaceKeyFunc(podPrefix, obj) },
-			GetAttrsFunc:   getPodAttrs,
-			NewFunc:        newFunc,
-			NewListFunc:    newListFunc,
-			Codec:          sc.Codec,
+			Storage:             s,
+			Versioner:           storage.APIObjectVersioner{},
+			GroupResource:       schema.GroupResource{Resource: "pods"},
+			EventsHistoryWindow: cacherstorage.DefaultEventFreshDuration,
+			ResourcePrefix:      podPrefix,
+			KeyFunc:             func(obj runtime.Object) (string, error) { return storage.NoNamespaceKeyFunc(podPrefix, obj) },
+			GetAttrsFunc:        getPodAttrs,
+			NewFunc:             newFunc,
+			NewListFunc:         newListFunc,
+			Codec:               sc.Codec,
 		}
 		cacher, err := cacherstorage.NewCacherFromConfig(config)
 		if err != nil {

staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go

@@ -61,10 +61,16 @@ const (
 	// storageWatchListPageSize is the cacher's request chunk size of
 	// initial and resync watch lists to storage.
 	storageWatchListPageSize = int64(10000)
+
+	// DefaultEventFreshDuration is the default time duration of events
+	// we want to keep.
+	// We set it to defaultBookmarkFrequency plus epsilon to maximize
+	// chances that last bookmark was sent within kept history, at the
+	// same time, minimizing the needed memory usage.
+	DefaultEventFreshDuration = defaultBookmarkFrequency + 15*time.Second
+
 	// defaultBookmarkFrequency defines how frequently watch bookmarks should be send
 	// in addition to sending a bookmark right before watch deadline.
-	//
-	// NOTE: Update `eventFreshDuration` when changing this value.
 	defaultBookmarkFrequency = time.Minute
 )
 
@@ -80,6 +86,10 @@ type Config struct {
 	// and metrics.
 	GroupResource schema.GroupResource
 
+	// EventsHistoryWindow specifies minimum history duration that storage is keeping.
+	// If lower than DefaultEventFreshDuration, the cache creation will fail.
+	EventsHistoryWindow time.Duration
+
 	// The Cache will be caching objects of a given Type and assumes that they
 	// are all stored under ResourcePrefix directory in the underlying database.
 	ResourcePrefix string
@@ -409,9 +419,15 @@ func NewCacherFromConfig(config Config) (*Cacher, error) {
 		contextMetadata = metadata.New(map[string]string{"source": "cache"})
 	}
 
+	eventFreshDuration := config.EventsHistoryWindow
+	if eventFreshDuration < DefaultEventFreshDuration {
+		return nil, fmt.Errorf("config.EventsHistoryWindow (%v) must not be below %v", eventFreshDuration, DefaultEventFreshDuration)
+	}
+
 	progressRequester := newConditionalProgressRequester(config.Storage.RequestWatchProgress, config.Clock, contextMetadata)
 	watchCache := newWatchCache(
-		config.KeyFunc, cacher.processEvent, config.GetAttrsFunc, config.Versioner, config.Indexers, config.Clock, config.GroupResource, progressRequester)
+		config.KeyFunc, cacher.processEvent, config.GetAttrsFunc, config.Versioner, config.Indexers,
+		config.Clock, eventFreshDuration, config.GroupResource, progressRequester)
 	listerWatcher := NewListerWatcher(config.Storage, config.ResourcePrefix, config.NewListFunc, contextMetadata)
 	reflectorName := "storage/cacher.go:" + config.ResourcePrefix
 

staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_test.go

@@ -470,18 +470,19 @@ func testSetupWithEtcdServer(t testing.TB, opts ...setupOption) (context.Context
 	}
 
 	config := Config{
-		Storage:        wrappedStorage,
-		Versioner:      storage.APIObjectVersioner{},
-		GroupResource:  schema.GroupResource{Resource: "pods"},
-		ResourcePrefix: setupOpts.resourcePrefix,
-		KeyFunc:        setupOpts.keyFunc,
-		GetAttrsFunc:   GetPodAttrs,
-		NewFunc:        newPod,
-		NewListFunc:    newPodList,
-		IndexerFuncs:   setupOpts.indexerFuncs,
-		Indexers:       &setupOpts.indexers,
-		Codec:          codecs.LegacyCodec(examplev1.SchemeGroupVersion),
-		Clock:          setupOpts.clock,
+		Storage:             wrappedStorage,
+		Versioner:           storage.APIObjectVersioner{},
+		GroupResource:       schema.GroupResource{Resource: "pods"},
+		EventsHistoryWindow: DefaultEventFreshDuration,
+		ResourcePrefix:      setupOpts.resourcePrefix,
+		KeyFunc:             setupOpts.keyFunc,
+		GetAttrsFunc:        GetPodAttrs,
+		NewFunc:             newPod,
+		NewListFunc:         newPodList,
+		IndexerFuncs:        setupOpts.indexerFuncs,
+		Indexers:            &setupOpts.indexers,
+		Codec:               codecs.LegacyCodec(examplev1.SchemeGroupVersion),
+		Clock:               setupOpts.clock,
 	}
 	cacher, err := NewCacherFromConfig(config)
 	if err != nil {

staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_whitebox_test.go

@@ -62,11 +62,12 @@ import (
 func newTestCacherWithoutSyncing(s storage.Interface) (*Cacher, storage.Versioner, error) {
 	prefix := "pods"
 	config := Config{
-		Storage:        s,
-		Versioner:      storage.APIObjectVersioner{},
-		GroupResource:  schema.GroupResource{Resource: "pods"},
-		ResourcePrefix: prefix,
-		KeyFunc:        func(obj runtime.Object) (string, error) { return storage.NamespaceKeyFunc(prefix, obj) },
+		Storage:             s,
+		Versioner:           storage.APIObjectVersioner{},
+		GroupResource:       schema.GroupResource{Resource: "pods"},
+		EventsHistoryWindow: DefaultEventFreshDuration,
+		ResourcePrefix:      prefix,
+		KeyFunc:             func(obj runtime.Object) (string, error) { return storage.NamespaceKeyFunc(prefix, obj) },
 		GetAttrsFunc: func(obj runtime.Object) (labels.Set, fields.Set, error) {
 			pod, ok := obj.(*example.Pod)
 			if !ok {
@@ -2722,17 +2723,18 @@ func TestWatchStreamSeparation(t *testing.T) {
 	setupOpts := &setupOptions{}
 	withDefaults(setupOpts)
 	config := Config{
-		Storage:        etcdStorage,
-		Versioner:      storage.APIObjectVersioner{},
-		GroupResource:  schema.GroupResource{Resource: "pods"},
-		ResourcePrefix: setupOpts.resourcePrefix,
-		KeyFunc:        setupOpts.keyFunc,
-		GetAttrsFunc:   GetPodAttrs,
-		NewFunc:        newPod,
-		NewListFunc:    newPodList,
-		IndexerFuncs:   setupOpts.indexerFuncs,
-		Codec:          codecs.LegacyCodec(examplev1.SchemeGroupVersion),
-		Clock:          setupOpts.clock,
+		Storage:             etcdStorage,
+		Versioner:           storage.APIObjectVersioner{},
+		GroupResource:       schema.GroupResource{Resource: "pods"},
+		EventsHistoryWindow: DefaultEventFreshDuration,
+		ResourcePrefix:      setupOpts.resourcePrefix,
+		KeyFunc:             setupOpts.keyFunc,
+		GetAttrsFunc:        GetPodAttrs,
+		NewFunc:             newPod,
+		NewListFunc:         newPodList,
+		IndexerFuncs:        setupOpts.indexerFuncs,
+		Codec:               codecs.LegacyCodec(examplev1.SchemeGroupVersion),
+		Clock:               setupOpts.clock,
 	}
 	tcs := []struct {
 		name                         string

staging/src/k8s.io/apiserver/pkg/storage/cacher/watch_cache.go

@@ -52,17 +52,11 @@ const (
 	// after receiving a 'too high resource version' error.
 	resourceVersionTooHighRetrySeconds = 1
 
-	// eventFreshDuration is time duration of events we want to keep.
-	// We set it to `defaultBookmarkFrequency` plus epsilon to maximize
-	// chances that last bookmark was sent within kept history, at the
-	// same time, minimizing the needed memory usage.
-	eventFreshDuration = 75 * time.Second
-
 	// defaultLowerBoundCapacity is a default value for event cache capacity's lower bound.
 	// TODO: Figure out, to what value we can decreased it.
 	defaultLowerBoundCapacity = 100
 
-	// defaultUpperBoundCapacity  should be able to keep eventFreshDuration of history.
+	// defaultUpperBoundCapacity should be able to keep the required history.
 	defaultUpperBoundCapacity = 100 * 1024
 )
 
@@ -142,6 +136,9 @@ type watchCache struct {
 	// for testing timeouts.
 	clock clock.Clock
 
+	// eventFreshDuration defines the minimum watch history watchcache will store.
+	eventFreshDuration time.Duration
+
 	// An underlying storage.Versioner.
 	versioner storage.Versioner
 
@@ -163,6 +160,7 @@ func newWatchCache(
 	versioner storage.Versioner,
 	indexers *cache.Indexers,
 	clock clock.WithTicker,
+	eventFreshDuration time.Duration,
 	groupResource schema.GroupResource,
 	progressRequester *conditionalProgressRequester) *watchCache {
 	wc := &watchCache{
@@ -179,6 +177,7 @@ func newWatchCache(
 		listResourceVersion: 0,
 		eventHandler:        eventHandler,
 		clock:               clock,
+		eventFreshDuration:  eventFreshDuration,
 		versioner:           versioner,
 		groupResource:       groupResource,
 		waitingUntilFresh:   progressRequester,
@@ -319,14 +318,14 @@ func (w *watchCache) updateCache(event *watchCacheEvent) {
 // - increases capacity by 2x if cache is full and all cached events occurred within last eventFreshDuration.
 // - decreases capacity by 2x when recent quarter of events occurred outside of eventFreshDuration(protect watchCache from flapping).
 func (w *watchCache) resizeCacheLocked(eventTime time.Time) {
-	if w.isCacheFullLocked() && eventTime.Sub(w.cache[w.startIndex%w.capacity].RecordTime) < eventFreshDuration {
+	if w.isCacheFullLocked() && eventTime.Sub(w.cache[w.startIndex%w.capacity].RecordTime) < w.eventFreshDuration {
 		capacity := min(w.capacity*2, w.upperBoundCapacity)
 		if capacity > w.capacity {
 			w.doCacheResizeLocked(capacity)
 		}
 		return
 	}
-	if w.isCacheFullLocked() && eventTime.Sub(w.cache[(w.endIndex-w.capacity/4)%w.capacity].RecordTime) > eventFreshDuration {
+	if w.isCacheFullLocked() && eventTime.Sub(w.cache[(w.endIndex-w.capacity/4)%w.capacity].RecordTime) > w.eventFreshDuration {
 		capacity := max(w.capacity/2, w.lowerBoundCapacity)
 		if capacity < w.capacity {
 			w.doCacheResizeLocked(capacity)
@@ -660,7 +659,7 @@ func (w *watchCache) suggestedWatchChannelSize(indexExists, triggerUsed bool) in
 	// We don't have an exact data, but given we store updates from
 	// the last <eventFreshDuration>, we approach it by dividing the
 	// capacity by the length of the history window.
-	chanSize := int(math.Ceil(float64(w.currentCapacity()) / eventFreshDuration.Seconds()))
+	chanSize := int(math.Ceil(float64(w.currentCapacity()) / w.eventFreshDuration.Seconds()))
 
 	// Finally we adjust the size to avoid ending with too low or
 	// to large values.