diff --git a/staging/src/k8s.io/apimachinery/pkg/apis/meta/internalversion/conversion.go b/staging/src/k8s.io/apimachinery/pkg/apis/meta/internalversion/conversion.go index 79020b277ce..f431157ec4e 100644 --- a/staging/src/k8s.io/apimachinery/pkg/apis/meta/internalversion/conversion.go +++ b/staging/src/k8s.io/apimachinery/pkg/apis/meta/internalversion/conversion.go @@ -17,43 +17,14 @@ limitations under the License. package internalversion import ( - "fmt" - v1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/conversion" - "k8s.io/apimachinery/pkg/sharding" ) -// Convert_v1_ListOptions_To_internalversion_ListOptions handles conversion from -// the wire-format v1.ListOptions to the internal ListOptions, including shard -// selector parsing. func Convert_v1_ListOptions_To_internalversion_ListOptions(in *v1.ListOptions, out *ListOptions, s conversion.Scope) error { - if err := autoConvert_v1_ListOptions_To_internalversion_ListOptions(in, out, s); err != nil { - return err - } - - // Parse the new shardSelector field into a ShardSelector if set. - if in.ShardSelector != "" { - sel, err := sharding.Parse(in.ShardSelector) - if err != nil { - return fmt.Errorf("invalid shard selector: %w", err) - } - out.ShardSelector = sel - } - - return nil + return autoConvert_v1_ListOptions_To_internalversion_ListOptions(in, out, s) } -// Convert_internalversion_ListOptions_To_v1_ListOptions handles conversion from -// internal ListOptions to the wire-format v1.ListOptions. func Convert_internalversion_ListOptions_To_v1_ListOptions(in *ListOptions, out *v1.ListOptions, s conversion.Scope) error { - if err := autoConvert_internalversion_ListOptions_To_v1_ListOptions(in, out, s); err != nil { - return err - } - - if in.ShardSelector != nil && !in.ShardSelector.Empty() { - out.ShardSelector = in.ShardSelector.String() - } - - return nil + return autoConvert_internalversion_ListOptions_To_v1_ListOptions(in, out, s) } diff --git a/staging/src/k8s.io/apimachinery/pkg/apis/meta/internalversion/types.go b/staging/src/k8s.io/apimachinery/pkg/apis/meta/internalversion/types.go index 3a58d055e3b..a34a11c3e30 100644 --- a/staging/src/k8s.io/apimachinery/pkg/apis/meta/internalversion/types.go +++ b/staging/src/k8s.io/apimachinery/pkg/apis/meta/internalversion/types.go @@ -21,7 +21,6 @@ import ( "k8s.io/apimachinery/pkg/fields" "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/runtime" - "k8s.io/apimachinery/pkg/sharding" ) // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object @@ -91,8 +90,10 @@ type ListOptions struct { // compatibility reasons) and to false otherwise. SendInitialEvents *bool - // ShardSelector is the parsed shard selector from the request. - ShardSelector sharding.Selector + // ShardSelector is the raw shard selector string from the request. + // Parsing is deferred to the apiserver storage layer where the CEL + // parser dependency is available. + ShardSelector string } // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object diff --git a/staging/src/k8s.io/apimachinery/pkg/sharding/parser.go b/staging/src/k8s.io/apimachinery/pkg/sharding/parser.go deleted file mode 100644 index c11b8681a21..00000000000 --- a/staging/src/k8s.io/apimachinery/pkg/sharding/parser.go +++ /dev/null @@ -1,42 +0,0 @@ -/* -Copyright The Kubernetes Authors. - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - -package sharding - -import "fmt" - -// ParseFunc is the signature for shard selector parsers. -type ParseFunc func(string) (Selector, error) - -var registeredParser ParseFunc - -// RegisterParser registers the shard selector parser implementation. -// This is called by k8s.io/apiserver to inject the CEL-based parser. -// -// This registration mechanism is an alpha-level internal API and is subject -// to change or removal in future releases. Do not depend on it. -func RegisterParser(p ParseFunc) { - registeredParser = p -} - -// Parse parses a shard selector string into a Selector. -// A parser must be registered via RegisterParser before calling Parse. -func Parse(s string) (Selector, error) { - if registeredParser == nil { - return nil, fmt.Errorf("no shard selector parser registered") - } - return registeredParser(s) -} diff --git a/staging/src/k8s.io/apimachinery/pkg/sharding/parser_test.go b/staging/src/k8s.io/apimachinery/pkg/sharding/parser_test.go deleted file mode 100644 index bc90ea2b941..00000000000 --- a/staging/src/k8s.io/apimachinery/pkg/sharding/parser_test.go +++ /dev/null @@ -1,59 +0,0 @@ -/* -Copyright The Kubernetes Authors. - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - -package sharding - -import ( - "testing" -) - -func TestParseWithoutRegisteredParser(t *testing.T) { - old := registeredParser - defer func() { registeredParser = old }() - - registeredParser = nil - _, err := Parse("anything") - if err == nil { - t.Error("Parse() without registered parser should return error") - } -} - -func TestRegisterParserAndParse(t *testing.T) { - old := registeredParser - defer func() { registeredParser = old }() - - called := false - RegisterParser(func(s string) (Selector, error) { - called = true - return NewSelector(ShardRangeRequirement{ - Key: "object.metadata.uid", - Start: "0x0", - End: "0x8", - }), nil - }) - - sel, err := Parse("test") - if err != nil { - t.Fatalf("unexpected error: %v", err) - } - if !called { - t.Error("registered parser was not called") - } - reqs := sel.Requirements() - if len(reqs) != 1 || reqs[0].Key != "object.metadata.uid" { - t.Errorf("unexpected requirements: %+v", reqs) - } -} diff --git a/staging/src/k8s.io/apimachinery/pkg/sharding/selector.go b/staging/src/k8s.io/apimachinery/pkg/sharding/selector.go index a454f263f16..841388a0d71 100644 --- a/staging/src/k8s.io/apimachinery/pkg/sharding/selector.go +++ b/staging/src/k8s.io/apimachinery/pkg/sharding/selector.go @@ -66,26 +66,34 @@ func (s *shardSelector) Matches(obj runtime.Object) (bool, error) { if len(s.requirements) == 0 { return true, nil } - // All requirements share the same key (enforced by the parser), - // so resolve the field value and compute the hash once. - value, err := ResolveFieldValue(obj, s.requirements[0].Key) + // All requirements must share the same key so we resolve the field value + // and compute the hash once. The parser enforces this, but we verify here + // to guard against selectors constructed through other means. + key := s.requirements[0].Key + for _, req := range s.requirements[1:] { + if req.Key != key { + return false, fmt.Errorf("inconsistent shard keys: %q vs %q", key, req.Key) + } + } + + value, err := ResolveFieldValue(obj, key) if err != nil { return false, err } hash := "0x" + HashField(value) for _, req := range s.requirements { - if !hexLess(hash, req.Start) && hexLess(hash, req.End) { + if !HexLess(hash, req.Start) && HexLess(hash, req.End) { return true, nil } } return false, nil } -// hexLess compares two lowercase hex strings numerically. -// It handles strings of different lengths by treating shorter strings -// as having smaller magnitude (fewer digits = smaller number). -func hexLess(a, b string) bool { +// HexLess compares two 0x-prefixed lowercase hex strings numerically. +// Both values must be normalized to 16 hex digits (e.g. "0x0000000000000000"), +// except for the special upper bound "0x10000000000000000" (2^64) which has 17. +func HexLess(a, b string) bool { if len(a) != len(b) { return len(a) < len(b) } @@ -117,6 +125,7 @@ func (s *shardSelector) DeepCopySelector() Selector { } // NewSelector creates a Selector from the given requirements. +// All requirements must use the same Key; this is validated at match time. // If no requirements are provided, returns Everything(). func NewSelector(reqs ...ShardRangeRequirement) Selector { if len(reqs) == 0 { diff --git a/staging/src/k8s.io/apiserver/pkg/registry/generic/registry/store.go b/staging/src/k8s.io/apiserver/pkg/registry/generic/registry/store.go index 2700c3a6e5b..4fef3be3975 100644 --- a/staging/src/k8s.io/apiserver/pkg/registry/generic/registry/store.go +++ b/staging/src/k8s.io/apiserver/pkg/registry/generic/registry/store.go @@ -45,6 +45,7 @@ import ( "k8s.io/apiserver/pkg/features" "k8s.io/apiserver/pkg/registry/generic" "k8s.io/apiserver/pkg/registry/rest" + "k8s.io/apiserver/pkg/sharding" "k8s.io/apiserver/pkg/storage" storeerr "k8s.io/apiserver/pkg/storage/errors" "k8s.io/apiserver/pkg/storage/etcd3/metrics" @@ -53,8 +54,6 @@ import ( flowcontrolrequest "k8s.io/apiserver/pkg/util/flowcontrol/request" "k8s.io/client-go/tools/cache" - _ "k8s.io/apiserver/pkg/sharding" // registers CEL-based shard selector parser - "k8s.io/klog/v2" ) @@ -395,8 +394,12 @@ func (e *Store) ListPredicate(ctx context.Context, p storage.SelectionPredicate, } p.Limit = options.Limit p.Continue = options.Continue - if utilfeature.DefaultFeatureGate.Enabled(features.ShardedListAndWatch) { - p.ShardSelector = options.ShardSelector + if utilfeature.DefaultFeatureGate.Enabled(features.ShardedListAndWatch) && options.ShardSelector != "" { + sel, err := sharding.Parse(options.ShardSelector) + if err != nil { + return nil, fmt.Errorf("invalid shard selector: %w", err) + } + p.ShardSelector = sel } list := e.NewListFunc() qualifiedResource := e.qualifiedResourceFromContext(ctx) @@ -1434,8 +1437,12 @@ func (e *Store) Watch(ctx context.Context, options *metainternalversion.ListOpti if options != nil { resourceVersion = options.ResourceVersion predicate.AllowWatchBookmarks = options.AllowWatchBookmarks - if utilfeature.DefaultFeatureGate.Enabled(features.ShardedListAndWatch) { - predicate.ShardSelector = options.ShardSelector + if utilfeature.DefaultFeatureGate.Enabled(features.ShardedListAndWatch) && options.ShardSelector != "" { + sel, err := sharding.Parse(options.ShardSelector) + if err != nil { + return nil, fmt.Errorf("invalid shard selector: %w", err) + } + predicate.ShardSelector = sel } } return e.WatchPredicate(ctx, predicate, resourceVersion, options.SendInitialEvents) diff --git a/staging/src/k8s.io/apiserver/pkg/sharding/parser.go b/staging/src/k8s.io/apiserver/pkg/sharding/parser.go index a7ac0dda309..fe08eac0c9b 100644 --- a/staging/src/k8s.io/apiserver/pkg/sharding/parser.go +++ b/staging/src/k8s.io/apiserver/pkg/sharding/parser.go @@ -142,10 +142,8 @@ func parseShardRangeCall(call ast.CallExpr) (apisharding.ShardRangeRequirement, return apisharding.ShardRangeRequirement{}, err } - // Validate start < end. Hex values are canonical (16 or 17 digits). - // A 17-digit value (only 0x10000000000000000) is always greater than a 16-digit value, - // so compare lengths first, then lexicographically within the same length. - if len(hexStart) > len(hexEnd) || (len(hexStart) == len(hexEnd) && hexStart >= hexEnd) { + // Validate start < end using the canonical hex comparison. + if !apisharding.HexLess(hexStart, hexEnd) { return apisharding.ShardRangeRequirement{}, fmt.Errorf("shard range start %s must be less than end %s", hexStart, hexEnd) } diff --git a/staging/src/k8s.io/apiserver/pkg/sharding/register.go b/staging/src/k8s.io/apiserver/pkg/sharding/register.go deleted file mode 100644 index 31c47b5ab7d..00000000000 --- a/staging/src/k8s.io/apiserver/pkg/sharding/register.go +++ /dev/null @@ -1,23 +0,0 @@ -/* -Copyright The Kubernetes Authors. - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - -package sharding - -import apisharding "k8s.io/apimachinery/pkg/sharding" - -func init() { - apisharding.RegisterParser(Parse) -} diff --git a/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go b/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go index 1923984f22e..326ed27c80d 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go @@ -677,8 +677,7 @@ func (c *Cacher) Watch(ctx context.Context, key string, opts storage.ListOptions return newImmediateCloseWatcher(), nil } - isSharded := pred.ShardSelector != nil && !pred.ShardSelector.Empty() - if isSharded { + if utilfeature.DefaultFeatureGate.Enabled(features.ShardedListAndWatch) && pred.ShardSelector != nil && !pred.ShardSelector.Empty() { metrics.RecordShardedWatchStarted(c.groupResource) originalForget := watcher.forget watcher.forget = func(drainWatcher bool) { @@ -805,9 +804,13 @@ func (c *Cacher) GetList(ctx context.Context, key string, opts storage.ListOptio if !ok { return fmt.Errorf("non *store.Element returned from storage: %v", obj) } - shardMatch, err := opts.Predicate.MatchesSharding(elem.Object) - if err != nil { - return fmt.Errorf("shard matching failed: %w", err) + shardMatch := true + if utilfeature.DefaultFeatureGate.Enabled(features.ShardedListAndWatch) { + var err error + shardMatch, err = opts.Predicate.MatchesSharding(elem.Object) + if err != nil { + return fmt.Errorf("shard matching failed: %w", err) + } } if shardMatch && opts.Predicate.MatchesObjectAttributes(elem.Labels, elem.Fields) { selectedObjects = append(selectedObjects, elem.Object) @@ -840,7 +843,9 @@ func (c *Cacher) GetList(ctx context.Context, key string, opts storage.ListOptio return err } } - opts.Predicate.SetShardInfoOnList(listObj) + if utilfeature.DefaultFeatureGate.Enabled(features.ShardedListAndWatch) { + opts.Predicate.SetShardInfoOnList(listObj) + } metrics.RecordListCacheMetrics(c.groupResource, indexUsed, len(resp.Items), listVal.Len()) return nil } @@ -1220,21 +1225,21 @@ func forgetWatcher(c *Cacher, w *cacheWatcher, index int, scope namespacedName, } func filterWithAttrsAndPrefixFunction(key string, p storage.SelectionPredicate, groupResource schema.GroupResource) filterWithAttrsFunc { - isSharded := p.ShardSelector != nil && !p.ShardSelector.Empty() + isSharded := utilfeature.DefaultFeatureGate.Enabled(features.ShardedListAndWatch) && p.ShardSelector != nil && !p.ShardSelector.Empty() filterFunc := func(objKey string, label labels.Set, field fields.Set, obj runtime.Object) bool { if !hasPathPrefix(objKey, key) { return false } - matches, err := p.MatchesSharding(obj) - if err != nil { - utilruntime.HandleError(fmt.Errorf("shard matching failed for %v: %w", groupResource, err)) - return false - } - if !matches { - if isSharded { - metrics.RecordWatchFilteredEvent(groupResource) + if isSharded { + matches, err := p.MatchesSharding(obj) + if err != nil { + utilruntime.HandleError(fmt.Errorf("shard matching failed for %v: %w", groupResource, err)) + return false + } + if !matches { + metrics.RecordWatchFilteredEvent(groupResource) + return false } - return false } return p.MatchesObjectAttributes(label, field) } diff --git a/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_whitebox_test.go b/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_whitebox_test.go index fa8ea1011b0..8005bf1ae70 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_whitebox_test.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_whitebox_test.go @@ -3411,6 +3411,7 @@ func makeExamplePod(name, namespace, uid string, podLabels map[string]string) *e } func TestFilterWithAttrsAndPrefixFunction_ShardMatchAndLabels(t *testing.T) { + featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ShardedListAndWatch, true) uid := "test-uid-1" pod := makeExamplePod("pod1", "default", uid, map[string]string{"app": "web"}) gr := schema.GroupResource{Resource: "pods"} @@ -3429,6 +3430,7 @@ func TestFilterWithAttrsAndPrefixFunction_ShardMatchAndLabels(t *testing.T) { } func TestFilterWithAttrsAndPrefixFunction_ShardMatchButLabelMismatch(t *testing.T) { + featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ShardedListAndWatch, true) uid := "test-uid-2" pod := makeExamplePod("pod2", "default", uid, map[string]string{"app": "api"}) gr := schema.GroupResource{Resource: "pods"} @@ -3447,6 +3449,7 @@ func TestFilterWithAttrsAndPrefixFunction_ShardMatchButLabelMismatch(t *testing. } func TestFilterWithAttrsAndPrefixFunction_ShardMismatch(t *testing.T) { + featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ShardedListAndWatch, true) uid := "test-uid-3" pod := makeExamplePod("pod3", "default", uid, map[string]string{"app": "web"}) gr := schema.GroupResource{Resource: "pods"} @@ -3465,6 +3468,7 @@ func TestFilterWithAttrsAndPrefixFunction_ShardMismatch(t *testing.T) { } func TestFilterWithAttrsAndPrefixFunction_NilShardSelector(t *testing.T) { + featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ShardedListAndWatch, true) pod := makeExamplePod("pod4", "default", "uid-4", map[string]string{"app": "web"}) gr := schema.GroupResource{Resource: "pods"} @@ -3516,6 +3520,7 @@ func TestFilterWithAttrsAndPrefixFunction_WrongPrefix(t *testing.T) { } func TestFilterWithAttrsAndPrefixFunction_ShardErrorDropsEvent(t *testing.T) { + featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ShardedListAndWatch, true) obj := &runtime.Unknown{} gr := schema.GroupResource{Resource: "pods"} @@ -3537,6 +3542,7 @@ func TestFilterWithAttrsAndPrefixFunction_ShardErrorDropsEvent(t *testing.T) { } func TestFilterWithAttrsAndPrefixFunction_NamespaceSharding(t *testing.T) { + featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ShardedListAndWatch, true) ns := "my-namespace" pod := makeExamplePod("pod-ns", ns, "some-uid", nil) @@ -3562,6 +3568,7 @@ func TestFilterWithAttrsAndPrefixFunction_NamespaceSharding(t *testing.T) { } func TestFilterWithAttrsAndPrefixFunction_NamespaceShardingMismatch(t *testing.T) { + featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ShardedListAndWatch, true) ns := "other-namespace" pod := makeExamplePod("pod-ns2", ns, "some-uid-2", nil) diff --git a/staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go b/staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go index e55fdcde875..8384832335f 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go @@ -898,7 +898,9 @@ func (s *store) GetList(ctx context.Context, key string, opts storage.ListOption if err := s.versioner.UpdateList(listObj, uint64(withRev), continueValue, remainingItemCount); err != nil { return err } - opts.Predicate.SetShardInfoOnList(listObj) + if utilfeature.DefaultFeatureGate.Enabled(features.ShardedListAndWatch) { + opts.Predicate.SetShardInfoOnList(listObj) + } return nil } diff --git a/staging/src/k8s.io/apiserver/pkg/storage/selection_predicate.go b/staging/src/k8s.io/apiserver/pkg/storage/selection_predicate.go index 419375c5f0b..90e6d12624c 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/selection_predicate.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/selection_predicate.go @@ -26,6 +26,8 @@ import ( "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/sharding" "k8s.io/apiserver/pkg/endpoints/request" + "k8s.io/apiserver/pkg/features" + utilfeature "k8s.io/apiserver/pkg/util/feature" ) // AttrFunc returns label and field sets and the uninitialized flag for List or Watch to match. @@ -178,6 +180,9 @@ func (s *SelectionPredicate) MatcherIndex(ctx context.Context) []MatchValue { // MatchesSharding returns true if the given object matches the sharding configuration. // If ShardSelector is set and non-empty, it delegates to ShardSelector.Matches(). func (s *SelectionPredicate) MatchesSharding(obj runtime.Object) (bool, error) { + if !utilfeature.DefaultFeatureGate.Enabled(features.ShardedListAndWatch) { + return true, nil + } if s.ShardSelector != nil && !s.ShardSelector.Empty() { return s.ShardSelector.Matches(obj) } @@ -186,6 +191,9 @@ func (s *SelectionPredicate) MatchesSharding(obj runtime.Object) (bool, error) { // SetShardInfoOnList sets shard metadata on the list response if sharding is active. func (s *SelectionPredicate) SetShardInfoOnList(listObj runtime.Object) { + if !utilfeature.DefaultFeatureGate.Enabled(features.ShardedListAndWatch) { + return + } if s.ShardSelector != nil && !s.ShardSelector.Empty() { if setter, ok := listObj.(metav1.ShardedListInterface); ok { setter.SetShardInfo(&metav1.ShardInfo{Selector: s.ShardSelector.String()}) diff --git a/test/integration/apiserver/sharding_test.go b/test/integration/apiserver/sharding_test.go index a765db2f390..a0af028d296 100644 --- a/test/integration/apiserver/sharding_test.go +++ b/test/integration/apiserver/sharding_test.go @@ -17,6 +17,7 @@ limitations under the License. package apiserver import ( + "context" "fmt" "math/big" "testing" @@ -24,12 +25,17 @@ import ( v1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/sharding" "k8s.io/apimachinery/pkg/watch" "k8s.io/apiserver/pkg/features" utilfeature "k8s.io/apiserver/pkg/util/feature" + "k8s.io/client-go/dynamic" featuregatetesting "k8s.io/component-base/featuregate/testing" + "k8s.io/kubernetes/cmd/kube-apiserver/app/options" + "k8s.io/kubernetes/test/integration/etcd" "k8s.io/kubernetes/test/integration/framework" + "k8s.io/kubernetes/test/utils/ktesting" ) // calculateShardRange divides the 64-bit hash space evenly across shards and @@ -50,30 +56,24 @@ func calculateShardRange(index, total int) (start, end string) { } func shardSelectorString(index, total int) string { - start, end := calculateShardRange(index, total) - return fmt.Sprintf("shardRange(object.metadata.uid, '%s', '%s')", start, end) + return shardSelectorStringForField("object.metadata.uid", index, total) } -// hexLess compares two lowercase hex strings numerically, matching the -// server-side sharding comparison logic. -func hexLess(a, b string) bool { - if len(a) != len(b) { - return len(a) < len(b) - } - return a < b +func shardSelectorStringForField(field string, index, total int) string { + start, end := calculateShardRange(index, total) + return fmt.Sprintf("shardRange(%s, '%s', '%s')", field, start, end) } // objectInShard returns true if the object's UID hash falls within the given shard range. func objectInShard(uid string, index, total int) bool { + return valueInShard(uid, index, total) +} + +// valueInShard returns true if the hash of value falls within the given shard range. +func valueInShard(value string, index, total int) bool { start, end := calculateShardRange(index, total) - hash := "0x" + sharding.HashField(uid) - if hexLess(hash, start) { - return false - } - if !hexLess(hash, end) { - return false - } - return true + hash := "0x" + sharding.HashField(value) + return !sharding.HexLess(hash, start) && sharding.HexLess(hash, end) } func TestShardedList(t *testing.T) { @@ -174,6 +174,7 @@ func TestShardedWatch(t *testing.T) { // Create objects and track which shard should see each one. const numObjects = 10 expectedShard := make(map[string]int) // UID -> shard index + created := make([]*v1.ConfigMap, 0, numObjects) for range numObjects { cm, err := client.CoreV1().ConfigMaps(ns.Name).Create(ctx, &v1.ConfigMap{ @@ -184,6 +185,7 @@ func TestShardedWatch(t *testing.T) { if err != nil { t.Fatalf("failed to create configmap: %v", err) } + created = append(created, cm) for shard := range numShards { if objectInShard(string(cm.UID), shard, numShards) { expectedShard[string(cm.UID)] = shard @@ -192,54 +194,85 @@ func TestShardedWatch(t *testing.T) { } } - // Collect events from each watcher. - received := make([]map[string]bool, numShards) - for i := range received { - received[i] = make(map[string]bool) + // Multiplex all watcher channels into a single channel for easy consumption. + type shardEvent struct { + shard int + eventType watch.EventType + uid string } - - for shard := range numShards { - collectEvents(t, watchers[shard], received[shard], expectedShard) - } - - // Verify every object was seen by exactly the right shard. - for uid, expectedIdx := range expectedShard { - if !received[expectedIdx][uid] { - t.Errorf("UID %s: expected in shard %d but not received", uid, expectedIdx) - } - for other := range numShards { - if other != expectedIdx && received[other][uid] { - t.Errorf("UID %s: received in shard %d but expected only in shard %d", uid, other, expectedIdx) - } - } - } -} - -func collectEvents(t *testing.T, w watch.Interface, seen map[string]bool, expected map[string]int) { - t.Helper() - timeout := time.After(30 * time.Second) - remaining := 0 - for range expected { - remaining++ - } - // We only need events for UIDs in our expected set that belong to this watcher. - // But we don't know which watcher this is, so just collect all ADDED events until - // we've seen enough or timed out. - for { - select { - case evt, ok := <-w.ResultChan(): - if !ok { - return - } - if evt.Type == watch.Added { + eventCh := make(chan shardEvent, 100) + for shard, w := range watchers { + go func(shard int, w watch.Interface) { + for evt := range w.ResultChan() { cm, ok := evt.Object.(*v1.ConfigMap) if !ok { continue } - seen[string(cm.UID)] = true + eventCh <- shardEvent{shard: shard, eventType: evt.Type, uid: string(cm.UID)} + } + }(shard, w) + } + + // waitForEvents drains the multiplexed channel until count events of the + // given type are collected, returning per-shard UID sets. + waitForEvents := func(eventType watch.EventType, count int) []map[string]bool { + t.Helper() + perShard := make([]map[string]bool, numShards) + for i := range perShard { + perShard[i] = make(map[string]bool) + } + timeout := time.After(30 * time.Second) + collected := 0 + for collected < count { + select { + case evt := <-eventCh: + if evt.eventType != eventType { + continue + } + perShard[evt.shard][evt.uid] = true + collected++ + case <-timeout: + t.Fatalf("timed out waiting for %s events: got %d/%d", eventType, collected, count) + } + } + return perShard + } + + // Collect ADDED events. + added := waitForEvents(watch.Added, numObjects) + verifyShardEvents(t, "ADDED", added, expectedShard, numShards) + + // Update each object by setting an annotation, then collect MODIFIED events. + for _, cm := range created { + cm.Annotations = map[string]string{"updated": "true"} + if _, err := client.CoreV1().ConfigMaps(ns.Name).Update(ctx, cm, metav1.UpdateOptions{}); err != nil { + t.Fatalf("failed to update configmap %s: %v", cm.Name, err) + } + } + modified := waitForEvents(watch.Modified, numObjects) + verifyShardEvents(t, "MODIFIED", modified, expectedShard, numShards) + + // Delete each object, then collect DELETED events. + for _, cm := range created { + if err := client.CoreV1().ConfigMaps(ns.Name).Delete(ctx, cm.Name, metav1.DeleteOptions{}); err != nil { + t.Fatalf("failed to delete configmap %s: %v", cm.Name, err) + } + } + deleted := waitForEvents(watch.Deleted, numObjects) + verifyShardEvents(t, "DELETED", deleted, expectedShard, numShards) +} + +// verifyShardEvents checks that each UID was seen by exactly the expected shard. +func verifyShardEvents(t *testing.T, eventType string, perShard []map[string]bool, expectedShard map[string]int, numShards int) { + t.Helper() + for uid, expectedIdx := range expectedShard { + if !perShard[expectedIdx][uid] { + t.Errorf("%s: UID %s expected in shard %d but not received", eventType, uid, expectedIdx) + } + for other := range numShards { + if other != expectedIdx && perShard[other][uid] { + t.Errorf("%s: UID %s received in shard %d but expected only in shard %d", eventType, uid, other, expectedIdx) } - case <-timeout: - return } } } @@ -313,3 +346,166 @@ func TestShardedListComplete(t *testing.T) { t.Errorf("expected %d items, got %d", numObjects, len(list.Items)) } } + +func TestShardedListByNamespace(t *testing.T) { + featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ShardedListAndWatch, true) + + ctx, client, _, tearDownFn := setup(t) + defer tearDownFn() + + // Create multiple namespaces with one ConfigMap each. + const numNamespaces = 10 + type nsObj struct { + namespace string + uid string + } + var objects []nsObj + for i := range numNamespaces { + nsName := fmt.Sprintf("shard-ns-%d", i) + ns := framework.CreateNamespaceOrDie(client, nsName, t) + defer framework.DeleteNamespaceOrDie(client, ns, t) + + cm, err := client.CoreV1().ConfigMaps(nsName).Create(ctx, &v1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: "test"}, + }, metav1.CreateOptions{}) + if err != nil { + t.Fatalf("failed to create configmap in %s: %v", nsName, err) + } + objects = append(objects, nsObj{namespace: nsName, uid: string(cm.UID)}) + } + + const numShards = 3 + allFound := make(map[string]bool) // UID -> found + + for shard := range numShards { + selector := shardSelectorStringForField("object.metadata.namespace", shard, numShards) + // List across all namespaces. + list, err := client.CoreV1().ConfigMaps("").List(ctx, metav1.ListOptions{ + ShardSelector: selector, + }) + if err != nil { + t.Fatalf("shard %d: failed to list: %v", shard, err) + } + + if list.ShardInfo == nil { + t.Errorf("shard %d: expected shardInfo to be set", shard) + } + + for _, cm := range list.Items { + if !valueInShard(cm.Namespace, shard, numShards) { + t.Errorf("shard %d: object %s/%s namespace hash should not be in this shard", shard, cm.Namespace, cm.Name) + } + allFound[string(cm.UID)] = true + } + } + + // Every object we created must appear in exactly one shard. + for _, obj := range objects { + if !allFound[obj.uid] { + t.Errorf("object in namespace %s (UID %s) was not returned by any shard", obj.namespace, obj.uid) + } + } +} + +// TestShardedListAllResources verifies that sharding is wired for every API +// resource. It follows the same discovery+etcd pattern used by the dryrun +// integration tests. +func TestShardedListAllResources(t *testing.T) { + featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ShardedListAndWatch, true) + + ctx := ktesting.Init(t) + client, config, tearDownFn := framework.StartTestServer(ctx, t, framework.TestServerSetup{ + ModifyServerRunOptions: func(opts *options.ServerRunOptions) { + opts.Admission.GenericAdmission.DisablePlugins = []string{"ServiceAccount"} + }, + }) + defer tearDownFn() + + dynamicClient, err := dynamic.NewForConfig(config) + if err != nil { + t.Fatal(err) + } + + testNamespace := "shard-allresources" + if _, err := client.CoreV1().Namespaces().Create(ctx, &v1.Namespace{ + ObjectMeta: metav1.ObjectMeta{Name: testNamespace}, + }, metav1.CreateOptions{}); err != nil { + t.Fatal(err) + } + + storageData := etcd.GetEtcdStorageDataForNamespace(testNamespace) + + _, serverResources, err := client.Discovery().ServerGroupsAndResources() + if err != nil { + t.Fatalf("failed to get ServerGroupsAndResources: %v", err) + } + + for _, resourceToTest := range etcd.GetResources(t, serverResources) { + mapping := resourceToTest.Mapping + gvr := mapping.Resource + + testData, hasData := storageData[gvr] + if !hasData { + continue + } + + // Check that this resource supports list. + hasListVerb := false + for _, discoveryGroup := range serverResources { + for _, r := range discoveryGroup.APIResources { + gv, _ := schema.ParseGroupVersion(discoveryGroup.GroupVersion) + if gv.WithResource(r.Name) == gvr { + for _, verb := range r.Verbs { + if verb == "list" { + hasListVerb = true + } + } + } + } + } + if !hasListVerb { + continue + } + + t.Run(gvr.String(), func(t *testing.T) { + res, obj, err := etcd.JSONToUnstructured(testData.Stub, testNamespace, mapping, dynamicClient) + if err != nil { + t.Fatalf("failed to unmarshal stub: %v", err) + } + + created, err := res.Create(ctx, obj, metav1.CreateOptions{}) + if err != nil { + t.Fatalf("failed to create %s: %v", gvr, err) + } + uid := string(created.GetUID()) + + // List with 2 shards and verify the object appears in exactly one. + const numShards = 2 + foundInShard := -1 + for shard := range numShards { + selector := shardSelectorString(shard, numShards) + listResult, err := res.List(ctx, metav1.ListOptions{ + ShardSelector: selector, + }) + if err != nil { + t.Fatalf("shard %d: list failed for %s: %v", shard, gvr, err) + } + for _, item := range listResult.Items { + if string(item.GetUID()) == uid { + if foundInShard >= 0 { + t.Errorf("object %s appeared in shard %d and %d", uid, foundInShard, shard) + } + foundInShard = shard + } + } + } + if foundInShard < 0 { + t.Errorf("object %s not found in any shard for %s", uid, gvr) + } + + if err := res.Delete(context.TODO(), created.GetName(), *metav1.NewDeleteOptions(0)); err != nil { + t.Logf("cleanup delete failed for %s: %v", gvr, err) + } + }) + } +}