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 c834ecf6d7c..c2685eda42f 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go @@ -230,7 +230,7 @@ func (s *store) getResourceSizeEstimator() *resourceSizeEstimator { // Get implements storage.Interface.Get. func (s *store) Get(ctx context.Context, key string, opts storage.GetOptions, out runtime.Object) error { - preparedKey, err := s.prepareKey(key) + preparedKey, err := s.prepareKey(key, false) if err != nil { return err } @@ -266,7 +266,7 @@ func (s *store) Get(ctx context.Context, key string, opts storage.GetOptions, ou // Create implements storage.Interface.Create. func (s *store) Create(ctx context.Context, key string, obj, out runtime.Object, ttl uint64) error { - preparedKey, err := s.prepareKey(key) + preparedKey, err := s.prepareKey(key, false) if err != nil { return err } @@ -336,7 +336,7 @@ func (s *store) Create(ctx context.Context, key string, obj, out runtime.Object, func (s *store) Delete( ctx context.Context, key string, out runtime.Object, preconditions *storage.Preconditions, validateDeletion storage.ValidateObjectFunc, cachedExistingObject runtime.Object, opts storage.DeleteOptions) error { - preparedKey, err := s.prepareKey(key) + preparedKey, err := s.prepareKey(key, false) if err != nil { return err } @@ -457,7 +457,7 @@ func (s *store) conditionalDelete( func (s *store) GuaranteedUpdate( ctx context.Context, key string, destination runtime.Object, ignoreNotFound bool, preconditions *storage.Preconditions, tryUpdate storage.UpdateFunc, cachedExistingObject runtime.Object) error { - preparedKey, err := s.prepareKey(key) + preparedKey, err := s.prepareKey(key, false) if err != nil { return err } @@ -645,13 +645,10 @@ func (s *store) Stats(ctx context.Context) (storage.Stats, error) { // returning stats without resource size startTime := time.Now() - prefix, err := s.prepareKey(s.resourcePrefix) + prefix, err := s.prepareKey(s.resourcePrefix, true) if err != nil { return storage.Stats{}, err } - if !strings.HasSuffix(prefix, "/") { - prefix += "/" - } count, err := s.client.Kubernetes.Count(ctx, prefix, kubernetes.CountOptions{}) metrics.RecordEtcdRequest("listWithCount", s.groupResource, err, startTime) if err != nil { @@ -677,13 +674,10 @@ func (s *store) EnableResourceSizeEstimation(getKeys storage.KeysFunc) error { func (s *store) getKeys(ctx context.Context) ([]string, error) { startTime := time.Now() - prefix, err := s.prepareKey(s.resourcePrefix) + prefix, err := s.prepareKey(s.resourcePrefix, true) if err != nil { return nil, err } - if !strings.HasSuffix(prefix, "/") { - prefix += "/" - } resp, err := s.client.KV.Get(ctx, prefix, clientv3.WithPrefix(), clientv3.WithKeysOnly()) metrics.RecordEtcdRequest("listOnlyKeys", s.groupResource, err, startTime) if err != nil { @@ -734,7 +728,7 @@ func (s *store) GetCurrentResourceVersion(ctx context.Context) (uint64, error) { // GetList implements storage.Interface. func (s *store) GetList(ctx context.Context, key string, opts storage.ListOptions, listObj runtime.Object) error { - keyPrefix, err := s.prepareKey(key) + keyPrefix, err := s.prepareKey(key, opts.Recursive) if err != nil { return err } @@ -755,14 +749,6 @@ func (s *store) GetList(ctx context.Context, key string, opts storage.ListOption return fmt.Errorf("need ptr to slice: %v", err) } - // For recursive lists, we need to make sure the key ended with "/" so that we only - // get children "directories". e.g. if we have key "/a", "/a/b", "/ab", getting keys - // with prefix "/a" will return all three, while with prefix "/a/" will return only - // "/a/b" which is the correct answer. - if opts.Recursive && !strings.HasSuffix(keyPrefix, "/") { - keyPrefix += "/" - } - // set the appropriate clientv3 options to filter the returned data set limit := opts.Predicate.Limit paging := opts.Predicate.Limit > 0 @@ -971,7 +957,7 @@ func growSlice(v reflect.Value, maxCapacity int, sizes ...int) { // Watch implements storage.Interface.Watch. func (s *store) Watch(ctx context.Context, key string, opts storage.ListOptions) (watch.Interface, error) { - preparedKey, err := s.prepareKey(key) + preparedKey, err := s.prepareKey(key, opts.Recursive) if err != nil { return nil, err } @@ -1118,7 +1104,7 @@ func (s *store) validateMinimumResourceVersion(minimumResourceVersion string, ac return nil } -func (s *store) prepareKey(key string) (string, error) { +func (s *store) prepareKey(key string, recursive bool) (string, error) { if key == ".." || strings.HasPrefix(key, "../") || strings.HasSuffix(key, "/..") || @@ -1139,6 +1125,13 @@ func (s *store) prepareKey(key string) (string, error) { if key[0] == '/' { startIndex = 1 } + // For recursive requests, we need to make sure the key ended with "/" so that we only + // get children "directories". e.g. if we have key "/a", "/a/b", "/ab", getting keys + // with prefix "/a" will return all three, while with prefix "/a/" will return only + // "/a/b" which is the correct answer. + if recursive && !strings.HasSuffix(key, "/") { + key += "/" + } return s.pathPrefix + key[startIndex:], nil } diff --git a/staging/src/k8s.io/apiserver/pkg/storage/etcd3/store_test.go b/staging/src/k8s.io/apiserver/pkg/storage/etcd3/store_test.go index 9fd14dd47b9..d154b3f7856 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/etcd3/store_test.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/etcd3/store_test.go @@ -655,7 +655,7 @@ func testSetup(t testing.TB, opts ...setupOption) (context.Context, *store, *kub return ctx, store, client } -func TestValidateKey(t *testing.T) { +func TestPrepareKey(t *testing.T) { validKeys := []string{ "/foo/bar/baz/a.b.c/", "/foo", @@ -689,21 +689,46 @@ func TestValidateKey(t *testing.T) { ) _, store, _ := testSetup(t, withPrefix(pathPrefix)) - for _, key := range validKeys { - k, err := store.prepareKey(key) - if err != nil { - t.Errorf("key %q should be valid; unexpected error: %v", key, err) - } else if !strings.HasPrefix(k, expectPrefix) { - t.Errorf("key %q should have prefix %q", k, expectPrefix) + t.Run("Non-recursive", func(t *testing.T) { + for _, key := range validKeys { + preparedKey, err := store.prepareKey(key, false) + if err != nil { + t.Errorf("preparing key %q should be valid; unexpected error: %v", key, err) + continue + } + if !strings.HasPrefix(preparedKey, expectPrefix) { + t.Errorf("prepared key %q should have prefix %q", preparedKey, expectPrefix) + } + if !strings.HasSuffix(preparedKey, key) { + t.Errorf("Prepared non-recursive key %q should have suffix %q", preparedKey, key) + } } - } + }) - for _, key := range invalidKeys { - _, err := store.prepareKey(key) - if err == nil { - t.Errorf("key %q should be invalid", key) + t.Run("Recursive", func(t *testing.T) { + for _, key := range validKeys { + preparedKey, err := store.prepareKey(key, true) + if err != nil { + t.Errorf("preparing key %q should be valid; unexpected error: %v", key, err) + continue + } + if !strings.HasPrefix(preparedKey, expectPrefix) { + t.Errorf("prepared key %q should have prefix %q", preparedKey, expectPrefix) + } + if !strings.HasSuffix(preparedKey, "/") { + t.Errorf("Prepared non-recursive key %q should have suffix '/'", preparedKey) + } } - } + }) + + t.Run("Invalid", func(t *testing.T) { + for _, key := range invalidKeys { + _, err := store.prepareKey(key, false) + if err == nil { + t.Errorf("key %q should be invalid", key) + } + } + }) } func TestInvalidKeys(t *testing.T) { diff --git a/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher.go b/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher.go index 93e356446b6..cf16606bc51 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher.go @@ -104,7 +104,7 @@ type watchChan struct { // pred must be non-nil. Only if opts.Predicate matches the change, it will be returned. func (w *watcher) Watch(ctx context.Context, key string, rev int64, opts storage.ListOptions) (watch.Interface, error) { if opts.Recursive && !strings.HasSuffix(key, "/") { - key += "/" + return nil, fmt.Errorf(`recursive key needs to end with "/"`) } if opts.ProgressNotify && w.newFunc == nil { return nil, apierrors.NewInternalError(errors.New("progressNotify for watch is unsupported by the etcd storage because no newFunc was provided")) diff --git a/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher_test.go b/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher_test.go index ede981ce253..8321867217a 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher_test.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher_test.go @@ -227,7 +227,7 @@ func TestTooLargeResourceVersionErrorForWatchList(t *testing.T) { t.Fatalf("Unable to convert NewTooLargeResourceVersionError to apierrors.StatusError") } - w, err := store.watcher.Watch(ctx, "/abc", int64(102), requestOpts) + w, err := store.watcher.Watch(ctx, "/abc/", int64(102), requestOpts) if err != nil { t.Fatal(err) }