Skip to content

K
k3s
Unverified Commit 9716f585 authored by Kubernetes Submit Queue's avatar Kubernetes Submit Queue Committed by GitHub
Browse files
Options

Merge pull request #64150 from wojtek-t/fix_list_resource_version

Automatic merge from submit-queue (batch tested with PRs 64102, 63303, 64150, 63841). If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>. Fix incorrectly set resource version in List Fix : https://github.com/kubernetes/kubernetes/issues/64147
parents 70caf339 a3578c86
Hide whitespace changes
Inline Side-by-side
Showing with 137 additions and 15 deletions
staging/src/k8s.io/apiserver/pkg/storage/etcd/etcd_helper.go
View file @ 9716f585
...@@ -347,7 +347,10 @@ func (h *etcdHelper) GetToList(ctx context.Context, key string, resourceVersion ...@@ -347,7 +347,10 @@ func (h *etcdHelper) GetToList(ctx context.Context, key string, resourceVersion
metrics.RecordEtcdRequestLatency("get", getTypeName(listPtr), startTime) metrics.RecordEtcdRequestLatency("get", getTypeName(listPtr), startTime)
if err != nil { if err != nil {
if etcdutil.IsEtcdNotFound(err) { if etcdutil.IsEtcdNotFound(err) {
return nil if etcdErr, ok := err.(etcd.Error); ok {
return h.versioner.UpdateList(listObj, etcdErr.Index, "")
}
return fmt.Errorf("unexpected error from storage: %#v", err)
} }
return toStorageErr(err, key, 0) return toStorageErr(err, key, 0)
} }
......
staging/src/k8s.io/apiserver/pkg/storage/etcd/etcd_helper_test.go
View file @ 9716f585
...@@ -329,6 +329,66 @@ func TestGet(t *testing.T) { ...@@ -329,6 +329,66 @@ func TestGet(t *testing.T) {
} }
} }
func TestGetToList(t *testing.T) {
_, codecs := testScheme(t)
codec := apitesting.TestCodec(codecs, examplev1.SchemeGroupVersion)
server := etcdtesting.NewEtcdTestClientServer(t)
defer server.Terminate(t)
key := "/some/key"
helper := newEtcdHelper(server.Client, codec, etcdtest.PathPrefix())
storedObj := &example.Pod{}
if err := helper.Create(context.TODO(), key, &example.Pod{ObjectMeta: metav1.ObjectMeta{Name: "foo"}}, storedObj, 0); err != nil {
t.Errorf("Unexpected error %#v", err)
}
tests := []struct {
key string
pred storage.SelectionPredicate
expectedOut []*example.Pod
}{{ // test GetToList on existing key
key: key,
pred: storage.Everything,
expectedOut: []*example.Pod{storedObj},
}, { // test GetToList on non-existing key
key: "/non-existing",
pred: storage.Everything,
expectedOut: nil,
}, { // test GetToList with matching pod name
key: "/non-existing",
pred: storage.SelectionPredicate{
Label: labels.Everything(),
Field: fields.ParseSelectorOrDie("metadata.name!=" + storedObj.Name),
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, bool, error) {
pod := obj.(*example.Pod)
return nil, fields.Set{"metadata.name": pod.Name}, pod.Initializers != nil, nil
},
},
expectedOut: nil,
}}
for i, tt := range tests {
out := &example.PodList{}
err := helper.GetToList(context.TODO(), tt.key, "", tt.pred, out)
if err != nil {
t.Fatalf("GetToList failed: %v", err)
}
if len(out.ResourceVersion) == 0 {
t.Errorf("#%d: unset resourceVersion", i)
}
if len(out.Items) != len(tt.expectedOut) {
t.Errorf("#%d: length of list want=%d, get=%d", i, len(tt.expectedOut), len(out.Items))
continue
}
for j, wantPod := range tt.expectedOut {
getPod := &out.Items[j]
if !reflect.DeepEqual(wantPod, getPod) {
t.Errorf("#%d: pod want=%#v, get=%#v", i, wantPod, getPod)
}
}
}
}
func TestGetNotFoundErr(t *testing.T) { func TestGetNotFoundErr(t *testing.T) {
_, codecs := testScheme(t) _, codecs := testScheme(t)
codec := apitesting.TestCodec(codecs, schema.GroupVersion{Version: "v1"}) codec := apitesting.TestCodec(codecs, schema.GroupVersion{Version: "v1"})
......
staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go
View file @ 9716f585
...@@ -388,26 +388,26 @@ func (s *store) GetToList(ctx context.Context, key string, resourceVersion strin ...@@ -388,26 +388,26 @@ func (s *store) GetToList(ctx context.Context, key string, resourceVersion strin
if err != nil { if err != nil {
return err return err
} }
key = path.Join(s.pathPrefix, key)
getResp, err := s.client.KV.Get(ctx, key, s.getOps...)
if err != nil {
return err
}
if len(getResp.Kvs) == 0 {
return nil
}
data, _, err := s.transformer.TransformFromStorage(getResp.Kvs[0].Value, authenticatedDataString(key))
if err != nil {
return storage.NewInternalError(err.Error())
}
v, err := conversion.EnforcePtr(listPtr) v, err := conversion.EnforcePtr(listPtr)
if err != nil || v.Kind() != reflect.Slice { if err != nil || v.Kind() != reflect.Slice {
panic("need ptr to slice") panic("need ptr to slice")
} }
if err := appendListItem(v, data, uint64(getResp.Kvs[0].ModRevision), pred, s.codec, s.versioner); err != nil {
key = path.Join(s.pathPrefix, key)
getResp, err := s.client.KV.Get(ctx, key, s.getOps...)
if err != nil {
return err return err
} }
if len(getResp.Kvs) > 0 {
data, _, err := s.transformer.TransformFromStorage(getResp.Kvs[0].Value, authenticatedDataString(key))
if err != nil {
return storage.NewInternalError(err.Error())
}
if err := appendListItem(v, data, uint64(getResp.Kvs[0].ModRevision), pred, s.codec, s.versioner); err != nil {
return err
}
}
// update version with cluster level revision // update version with cluster level revision
return s.versioner.UpdateList(listObj, uint64(getResp.Header.Revision), "") return s.versioner.UpdateList(listObj, uint64(getResp.Header.Revision), "")
} }
......
staging/src/k8s.io/apiserver/pkg/storage/etcd3/store_test.go
View file @ 9716f585
...@@ -326,6 +326,9 @@ func TestGetToList(t *testing.T) { ...@@ -326,6 +326,9 @@ func TestGetToList(t *testing.T) {
if err != nil { if err != nil {
t.Fatalf("GetToList failed: %v", err) t.Fatalf("GetToList failed: %v", err)
} }
if len(out.ResourceVersion) == 0 {
t.Errorf("#%d: unset resourceVersion", i)
}
if len(out.Items) != len(tt.expectedOut) { if len(out.Items) != len(tt.expectedOut) {
t.Errorf("#%d: length of list want=%d, get=%d", i, len(tt.expectedOut), len(out.Items)) t.Errorf("#%d: length of list want=%d, get=%d", i, len(tt.expectedOut), len(out.Items))
continue continue
......
staging/src/k8s.io/apiserver/pkg/storage/tests/cacher_test.go
View file @ 9716f585
...@@ -166,6 +166,62 @@ func TestGet(t *testing.T) { ...@@ -166,6 +166,62 @@ func TestGet(t *testing.T) {
} }
} }
func TestGetToList(t *testing.T) {
server, etcdStorage := newEtcdTestStorage(t, etcdtest.PathPrefix())
defer server.Terminate(t)
cacher, _ := newTestCacher(etcdStorage, 10)
defer cacher.Stop()
storedObj := updatePod(t, etcdStorage, makeTestPod("foo"), nil)
key := "pods/" + storedObj.Namespace + "/" + storedObj.Name
tests := []struct {
key string
pred storage.SelectionPredicate
expectedOut []*example.Pod
}{{ // test GetToList on existing key
key: key,
pred: storage.Everything,
expectedOut: []*example.Pod{storedObj},
}, { // test GetToList on non-existing key
key: "/non-existing",
pred: storage.Everything,
expectedOut: nil,
}, { // test GetToList with matching pod name
key: "/non-existing",
pred: storage.SelectionPredicate{
Label: labels.Everything(),
Field: fields.ParseSelectorOrDie("metadata.name!=" + storedObj.Name),
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, bool, error) {
pod := obj.(*example.Pod)
return nil, fields.Set{"metadata.name": pod.Name}, pod.Initializers != nil, nil
},
},
expectedOut: nil,
}}
for i, tt := range tests {
out := &example.PodList{}
err := cacher.GetToList(context.TODO(), tt.key, "", tt.pred, out)
if err != nil {
t.Fatalf("GetToList failed: %v", err)
}
if len(out.ResourceVersion) == 0 {
t.Errorf("#%d: unset resourceVersion", i)
}
if len(out.Items) != len(tt.expectedOut) {
t.Errorf("#%d: length of list want=%d, get=%d", i, len(tt.expectedOut), len(out.Items))
continue
}
for j, wantPod := range tt.expectedOut {
getPod := &out.Items[j]
if !reflect.DeepEqual(wantPod, getPod) {
t.Errorf("#%d: pod want=%#v, get=%#v", i, wantPod, getPod)
}
}
}
}
func TestList(t *testing.T) { func TestList(t *testing.T) {
server, etcdStorage := newEtcdTestStorage(t, etcdtest.PathPrefix()) server, etcdStorage := newEtcdTestStorage(t, etcdtest.PathPrefix())
defer server.Terminate(t) defer server.Terminate(t)
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment