github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2026-07-18 04:54:54 +00:00
Code Issues Projects Releases Wiki Activity

sharding: address review comments (deads2k)

Browse Source
This commit is contained in:
Jefftree
2026-03-16 18:30:48 -04:00
parent d43dc1abd8
commit ab1cfa764e
13 changed files with 332 additions and 252 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
staging/src/k8s.io/apimachinery/pkg/apis/meta/internalversion/conversion.go
View File

@@ -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)
}

7
staging/src/k8s.io/apimachinery/pkg/apis/meta/internalversion/types.go
View File

@@ -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

42
staging/src/k8s.io/apimachinery/pkg/sharding/parser.go
View File

@@ -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)
}

59
staging/src/k8s.io/apimachinery/pkg/sharding/parser_test.go
View File

@@ -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)
}
}

25
staging/src/k8s.io/apimachinery/pkg/sharding/selector.go
View File

@@ -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 {

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

@@ -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)

6
staging/src/k8s.io/apiserver/pkg/sharding/parser.go
View File

@@ -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)
}

23
staging/src/k8s.io/apiserver/pkg/sharding/register.go
View File

@@ -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)
}

37
staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go
View File

@@ -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)
}

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

@@ -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)

4
staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go
View File

@@ -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
}

8
staging/src/k8s.io/apiserver/pkg/storage/selection_predicate.go
View File

@@ -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()})

314
test/integration/apiserver/sharding_test.go
View File

@@ -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)
}
})
}
}