kubernetes
/
apiserver
mirror of https://github.com/kubernetes/apiserver.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #130775 from serathius/watchcache-consistent-read 

Fix detecting consistent read when watchcache starts handling continue

Kubernetes-commit: 54e7d2760d1c941c8db50d059dbbe0cfb54a909a
This commit is contained in:
Kubernetes Publisher 2025-03-13 14:55:54 -07:00
parent d88392f1f6 3cb2448d98
commit 543428d1c4
2 changed files with 24 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
pkg/storage/cacher/delegator.go
View File

@ -176,7 +176,8 @@ func (c *CacheDelegator) Get(ctx context.Context, key string, opts storage.GetOp
}
func (c *CacheDelegator) GetList(ctx context.Context, key string, opts storage.ListOptions, listObj runtime.Object) error {
if shouldDelegateList(opts) {
shouldDelegate, consistentRead := shouldDelegateList(opts)
if shouldDelegate {
return c.storage.GetList(ctx, key, opts, listObj)
}
@ -198,8 +199,6 @@ func (c *CacheDelegator) GetList(ctx context.Context, key string, opts storage.L
return c.storage.GetList(ctx, key, opts, listObj)
}
}
requestWatchProgressSupported := etcdfeature.DefaultFeatureSupportChecker.Supports(storage.RequestWatchProgress)
consistentRead := opts.ResourceVersion == "" && utilfeature.DefaultFeatureGate.Enabled(features.ConsistentListFromCache) && requestWatchProgressSupported
if consistentRead {
listRV, err = c.storage.GetCurrentResourceVersion(ctx)
if err != nil {
@ -243,31 +242,33 @@ func shouldDelegateListOnNotReadyCache(opts storage.ListOptions) bool {
// NOTICE: Keep in sync with shouldListFromStorage function in
//
// staging/src/k8s.io/apiserver/pkg/util/flowcontrol/request/list_work_estimator.go
func shouldDelegateList(opts storage.ListOptions) bool {
func shouldDelegateList(opts storage.ListOptions) (shouldDeletage, consistentRead bool) {
// see https://kubernetes.io/docs/reference/using-api/api-concepts/#semantics-for-get-and-list
consistentRead = false
switch opts.ResourceVersionMatch {
case metav1.ResourceVersionMatchExact:
return true
return true, consistentRead
case metav1.ResourceVersionMatchNotOlderThan:
return false
return false, consistentRead
case "":
// Legacy exact match
if opts.Predicate.Limit > 0 && len(opts.ResourceVersion) > 0 && opts.ResourceVersion != "0" {
return true
return true, consistentRead
}
// Continue
if len(opts.Predicate.Continue) > 0 {
return true
return true, consistentRead
}
// Consistent Read
if opts.ResourceVersion == "" {
consistentRead = true
consistentListFromCacheEnabled := utilfeature.DefaultFeatureGate.Enabled(features.ConsistentListFromCache)
requestWatchProgressSupported := etcdfeature.DefaultFeatureSupportChecker.Supports(storage.RequestWatchProgress)
return !consistentListFromCacheEnabled || !requestWatchProgressSupported
return !consistentListFromCacheEnabled || !requestWatchProgressSupported, consistentRead
}
return false
return false, consistentRead
default:
return true
return true, consistentRead
}
}

23
pkg/util/flowcontrol/request/list_work_estimator.go
View File

@ -19,7 +19,6 @@ package request
import (
"math"
"net/http"
"net/url"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime/schema"
@ -86,8 +85,8 @@ func (e *listWorkEstimator) estimate(r *http.Request, flowSchemaName, priorityLe
return WorkEstimate{InitialSeats: e.config.MinimumSeats}
}
}
isListFromCache := requestInfo.Verb == "watch" || !shouldListFromStorage(query, &listOptions)
listFromStorage, _ := shouldListFromStorage(&listOptions)
isListFromCache := requestInfo.Verb == "watch" || !listFromStorage
numStored, err := e.countGetterFn(key(requestInfo))
switch {
@ -163,30 +162,32 @@ func key(requestInfo *apirequest.RequestInfo) string {
// NOTICE: Keep in sync with shouldDelegateList function in
//
// staging/src/k8s.io/apiserver/pkg/storage/cacher/delegator.go
func shouldListFromStorage(query url.Values, opts *metav1.ListOptions) bool {
func shouldListFromStorage(opts *metav1.ListOptions) (shouldDeletage, consistentRead bool) {
// see https://kubernetes.io/docs/reference/using-api/api-concepts/#semantics-for-get-and-list
consistentRead = false
switch opts.ResourceVersionMatch {
case metav1.ResourceVersionMatchExact:
return true
return true, consistentRead
case metav1.ResourceVersionMatchNotOlderThan:
return false
return false, consistentRead
case "":
// Legacy exact match
if opts.Limit > 0 && len(opts.ResourceVersion) > 0 && opts.ResourceVersion != "0" {
return true
return true, consistentRead
}
// Continue
if len(opts.Continue) > 0 {
return true
return true, consistentRead
}
// Consistent Read
if opts.ResourceVersion == "" {
consistentRead = true
consistentListFromCacheEnabled := utilfeature.DefaultFeatureGate.Enabled(features.ConsistentListFromCache)
requestWatchProgressSupported := etcdfeature.DefaultFeatureSupportChecker.Supports(storage.RequestWatchProgress)
return !consistentListFromCacheEnabled || !requestWatchProgressSupported
return !consistentListFromCacheEnabled || !requestWatchProgressSupported, consistentRead
}
return false
return false, consistentRead
default:
return true
return true, consistentRead
}
}