github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-23 19:56:01 +00:00
Code Issues Projects Releases Wiki Activity

use Watch for single object instead of WatchList

Browse Source
This commit is contained in:
Masahiro Sano 2015-04-18 01:23:58 +09:00
parent 23e806604d
commit f90dc8f413
6 changed files with 63 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
pkg/registry/etcd/etcd.go
View File

@ -271,7 +271,8 @@ func (r *Registry) WatchServices(ctx api.Context, label labels.Selector, field f
if err != nil {
return nil, err
}
return r.Watch(key, version), nil
// TODO: use generic.SelectionPredicate
return r.Watch(key, version, tools.Everything)
}
if field.Empty() {
return r.WatchList(makeServiceListKey(ctx), version, tools.Everything)

25
pkg/registry/generic/etcd/etcd.go
View File

@ -429,18 +429,7 @@ func (e *Etcd) WatchPredicate(ctx api.Context, m generic.Matcher, resourceVersio
return nil, err
}
var watchKey string
if name, ok := m.MatchesSingle(); ok {
key, err := e.KeyFunc(ctx, name)
if err != nil {
return nil, err
}
watchKey = key
} else {
watchKey = e.KeyRootFunc(ctx)
}
return e.Helper.WatchList(watchKey, version, func(obj runtime.Object) bool {
filterFunc := func(obj runtime.Object) bool {
matches, err := m.Matches(obj)
if err != nil {
glog.Errorf("unable to match watch: %v", err)
@ -453,5 +442,15 @@ func (e *Etcd) WatchPredicate(ctx api.Context, m generic.Matcher, resourceVersio
}
}
return matches
})
}
if name, ok := m.MatchesSingle(); ok {
key, err := e.KeyFunc(ctx, name)
if err != nil {
return nil, err
}
return e.Helper.Watch(key, version, filterFunc)
}
return e.Helper.WatchList(e.KeyRootFunc(ctx), version, filterFunc)
}

9
pkg/registry/generic/etcd/etcd_test.go
View File

@ -690,11 +690,16 @@ func TestEtcdWatch(t *testing.T) {
for name, m := range table {
podA := &api.Pod{
ObjectMeta: api.ObjectMeta{Name: "foo", ResourceVersion: "1"},
Spec: api.PodSpec{Host: "machine"},
ObjectMeta: api.ObjectMeta{
Name: "foo",
Namespace: api.NamespaceDefault,
ResourceVersion: "1",
},
Spec: api.PodSpec{Host: "machine"},
}
respWithPodA := &etcd.Response{
Node: &etcd.Node{
Key: "/registry/pods/default/foo",
Value: runtime.EncodeOrDie(testapi.Codec(), podA),
ModifiedIndex: 1,
CreatedIndex: 1,

6
pkg/tools/etcd_helper_watch.go
View File

@ -79,8 +79,10 @@ func (h *EtcdHelper) WatchList(key string, resourceVersion uint64, filter Filter
// Watch begins watching the specified key. Events are decoded into
// API objects and sent down the returned watch.Interface.
// Errors will be sent down the channel.
func (h *EtcdHelper) Watch(key string, resourceVersion uint64) watch.Interface {
return h.WatchAndTransform(key, resourceVersion, nil)
func (h *EtcdHelper) Watch(key string, resourceVersion uint64, filter FilterFunc) (watch.Interface, error) {
w := newEtcdWatcher(false, nil, filter, h.Codec, h.Versioner, nil)
go w.etcdWatch(h.Client, key, resourceVersion)
return w, nil
}
// WatchAndTransform begins watching the specified key. Events are decoded into

40
pkg/tools/etcd_helper_watch_test.go
View File

@ -207,7 +207,13 @@ func TestWatchEtcdError(t *testing.T) {
fakeClient.WatchImmediateError = fmt.Errorf("immediate error")
h := EtcdHelper{fakeClient, codec, versioner}
got := <-h.Watch("/some/key", 4).ResultChan()
watching, err := h.Watch("/some/key", 4, Everything)
if err != nil {
t.Fatalf("Unexpected error: %v", err)
}
defer watching.Stop()
got := <-watching.ResultChan()
if got.Type != watch.Error {
t.Fatalf("Unexpected non-error")
}
@ -229,7 +235,10 @@ func TestWatch(t *testing.T) {
fakeClient.expectNotFoundGetSet["/some/key"] = struct{}{}
h := EtcdHelper{fakeClient, codec, versioner}
watching := h.Watch("/some/key", 0)
watching, err := h.Watch("/some/key", 0, Everything)
if err != nil {
t.Fatalf("Unexpected error: %v", err)
}
fakeClient.WaitForWatchCompletion()
// when server returns not found, the watch index starts at the next value (1)
@ -398,7 +407,11 @@ func TestWatchEtcdState(t *testing.T) {
fakeClient.Data[key] = value
}
h := EtcdHelper{fakeClient, codec, versioner}
watching := h.Watch("/somekey/foo", testCase.From)
watching, err := h.Watch("/somekey/foo", testCase.From, Everything)
if err != nil {
t.Fatalf("Unexpected error: %v", err)
}
fakeClient.WaitForWatchCompletion()
t.Logf("Testing %v", k)
@ -466,7 +479,10 @@ func TestWatchFromZeroIndex(t *testing.T) {
fakeClient.Data["/some/key"] = testCase.Response
h := EtcdHelper{fakeClient, codec, versioner}
watching := h.Watch("/some/key", 0)
watching, err := h.Watch("/some/key", 0, Everything)
if err != nil {
t.Fatalf("Unexpected error: %v", err)
}
fakeClient.WaitForWatchCompletion()
if e, a := testCase.Response.R.EtcdIndex+1, fakeClient.WatchIndex; e != a {
@ -612,7 +628,10 @@ func TestWatchFromNotFound(t *testing.T) {
}
h := EtcdHelper{fakeClient, codec, versioner}
watching := h.Watch("/some/key", 0)
watching, err := h.Watch("/some/key", 0, Everything)
if err != nil {
t.Fatalf("Unexpected error: %v", err)
}
fakeClient.WaitForWatchCompletion()
if fakeClient.WatchIndex != 3 {
@ -635,7 +654,10 @@ func TestWatchFromOtherError(t *testing.T) {
}
h := EtcdHelper{fakeClient, codec, versioner}
watching := h.Watch("/some/key", 0)
watching, err := h.Watch("/some/key", 0, Everything)
if err != nil {
t.Fatalf("Unexpected error: %v", err)
}
errEvent := <-watching.ResultChan()
if e, a := watch.Error, errEvent.Type; e != a {
@ -665,7 +687,11 @@ func TestWatchPurposefulShutdown(t *testing.T) {
fakeClient.expectNotFoundGetSet["/some/key"] = struct{}{}
// Test purposeful shutdown
watching := h.Watch("/some/key", 0)
watching, err := h.Watch("/some/key", 0, Everything)
if err != nil {
t.Fatalf("Unexpected error: %v", err)
}
fakeClient.WaitForWatchCompletion()
watching.Stop()

6
test/integration/etcd_tools_test.go
View File

@ -101,7 +101,11 @@ func TestWatch(t *testing.T) {
expectedVersion := resp.Node.ModifiedIndex
// watch should load the object at the current index
w := helper.Watch(key, 0)
w, err := helper.Watch(key, 0, tools.Everything)
if err != nil {
t.Fatalf("Unexpected error: %v", err)
}
event := <-w.ResultChan()
if event.Type != watch.Added || event.Object == nil {
t.Fatalf("expected first value to be set to ADDED, got %#v", event)