Add generic registry object so we can stop rewriting this code

pkg/registry/generic/doc.go

@@ -0,0 +1,19 @@
+/*
+Copyright 2014 Google Inc. All rights reserved.
+
+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 generic provides a generic object store interface and a
+// generic label/field matching type.
+package generic

pkg/registry/generic/etcd.go

@@ -0,0 +1,119 @@
+/*
+Copyright 2014 Google Inc. All rights reserved.
+
+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 generic
+
+import (
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	etcderr "github.com/GoogleCloudPlatform/kubernetes/pkg/api/errors/etcd"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
+)
+
+// Etcd implements generic.Registry, backing it with etcd storage.
+// It's intended to be embeddable, so that you can implement any
+// non-generic functions if needed.
+// You must supply a value for every field below before use; these are
+// left public as it's meant to be overridable if need be.
+type Etcd struct {
+	// Called to make a new object, should return e.g., &api.Pod{}
+	NewFunc func() runtime.Object
+
+	// Called to make a new listing object, should return e.g., &api.PodList{}
+	NewListFunc func() runtime.Object
+
+	// Used for error reporting
+	EndpointName string
+
+	// Used for listing/watching; should not include trailing "/"
+	KeyRoot string
+
+	// Called for Create/Update/Get/Delete
+	KeyFunc func(id string) string
+
+	// Used for all etcd access functions
+	Helper tools.EtcdHelper
+}
+
+// List returns a list of all the items matching m.
+func (e *Etcd) List(ctx api.Context, m Matcher) (runtime.Object, error) {
+	list := e.NewListFunc()
+	err := e.Helper.ExtractToList(e.KeyRoot, list)
+	if err != nil {
+		return nil, err
+	}
+	return FilterList(list, m)
+}
+
+// FilterList filters any list object that conforms to the api conventions,
+// provided that 'm' works with the concrete type of list.
+func FilterList(list runtime.Object, m Matcher) (filtered runtime.Object, err error) {
+	// TODO: push a matcher down into tools.EtcdHelper to avoid all this
+	// nonsense. This is a lot of unnecessary copies.
+	items, err := runtime.ExtractList(list)
+	if err != nil {
+		return nil, err
+	}
+	var filteredItems []runtime.Object
+	for _, obj := range items {
+		if match, err := m.Matches(obj); err == nil && match {
+			filteredItems = append(filteredItems, obj)
+		}
+	}
+	err = runtime.SetList(list, filteredItems)
+	if err != nil {
+		return nil, err
+	}
+	return list, nil
+}
+
+// Create inserts a new item.
+func (e *Etcd) Create(ctx api.Context, id string, obj runtime.Object) error {
+	err := e.Helper.CreateObj(e.KeyFunc(id), obj, 0)
+	return etcderr.InterpretCreateError(err, e.EndpointName, id)
+}
+
+// Update updates the item.
+func (e *Etcd) Update(ctx api.Context, id string, obj runtime.Object) error {
+	err := e.Helper.SetObj(e.KeyFunc(id), obj)
+	return etcderr.InterpretUpdateError(err, e.EndpointName, id)
+}
+
+// Get retrieves the item from etcd.
+func (e *Etcd) Get(ctx api.Context, id string) (runtime.Object, error) {
+	obj := e.NewFunc()
+	err := e.Helper.ExtractObj(e.KeyFunc(id), obj, false)
+	if err != nil {
+		return nil, etcderr.InterpretGetError(err, e.EndpointName, id)
+	}
+	return obj, nil
+}
+
+// Delete removes the item from etcd.
+func (e *Etcd) Delete(ctx api.Context, id string) error {
+	err := e.Helper.Delete(e.KeyFunc(id), false)
+	return etcderr.InterpretDeleteError(err, e.EndpointName, id)
+}
+
+// Watch starts a watch for the items that m matches.
+// TODO: Detect if m references a single object instead of a list.
+func (e *Etcd) Watch(ctx api.Context, m Matcher, resourceVersion uint64) (watch.Interface, error) {
+	return e.Helper.WatchList(e.KeyRoot, resourceVersion, func(obj runtime.Object) bool {
+		matches, err := m.Matches(obj)
+		return err == nil && matches
+	})
+}

pkg/registry/generic/etcd_test.go

@@ -0,0 +1,437 @@
+/*
+Copyright 2014 Google Inc. All rights reserved.
+
+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 generic
+
+import (
+	"fmt"
+	"path"
+	"reflect"
+	"testing"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/errors"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/testapi"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
+
+	"github.com/coreos/go-etcd/etcd"
+)
+
+func NewTestGenericEtcdRegistry(t *testing.T) (*tools.FakeEtcdClient, *Etcd) {
+	f := tools.NewFakeEtcdClient(t)
+	f.TestIndex = true
+	h := tools.EtcdHelper{f, testapi.Codec(), tools.RuntimeVersionAdapter{testapi.ResourceVersioner()}}
+	return f, &Etcd{
+		NewFunc:      func() runtime.Object { return &api.Pod{} },
+		NewListFunc:  func() runtime.Object { return &api.PodList{} },
+		EndpointName: "pods",
+		KeyRoot:      "/registry/pods",
+		KeyFunc: func(id string) string {
+			return path.Join("/registry/pods", id)
+		},
+		Helper: h,
+	}
+}
+
+// SetMatcher is a matcher that matches any pod with id in the set.
+// Makes testing simpler.
+type SetMatcher struct {
+	util.StringSet
+}
+
+func (sm SetMatcher) Matches(obj runtime.Object) (bool, error) {
+	pod, ok := obj.(*api.Pod)
+	if !ok {
+		return false, fmt.Errorf("wrong object")
+	}
+	return sm.Has(pod.ID), nil
+}
+
+// EverythingMatcher matches everything
+type EverythingMatcher struct{}
+
+func (EverythingMatcher) Matches(obj runtime.Object) (bool, error) {
+	return true, nil
+}
+
+func TestEtcdList(t *testing.T) {
+	podA := &api.Pod{
+		TypeMeta:     api.TypeMeta{ID: "foo"},
+		DesiredState: api.PodState{Host: "machine"},
+	}
+	podB := &api.Pod{
+		TypeMeta:     api.TypeMeta{ID: "bar"},
+		DesiredState: api.PodState{Host: "machine"},
+	}
+
+	normalListResp := &etcd.Response{
+		Node: &etcd.Node{
+			Nodes: []*etcd.Node{
+				{Value: runtime.EncodeOrDie(testapi.Codec(), podA)},
+				{Value: runtime.EncodeOrDie(testapi.Codec(), podB)},
+			},
+		},
+	}
+
+	table := map[string]struct {
+		in      tools.EtcdResponseWithError
+		m       Matcher
+		out     runtime.Object
+		succeed bool
+	}{
+		"empty": {
+			in: tools.EtcdResponseWithError{
+				R: &etcd.Response{
+					Node: &etcd.Node{
+						Nodes: []*etcd.Node{},
+					},
+				},
+				E: nil,
+			},
+			m:       EverythingMatcher{},
+			out:     &api.PodList{Items: []api.Pod{}},
+			succeed: true,
+		},
+		"notFound": {
+			in: tools.EtcdResponseWithError{
+				R: &etcd.Response{},
+				E: tools.EtcdErrorNotFound,
+			},
+			m:       EverythingMatcher{},
+			out:     &api.PodList{Items: []api.Pod{}},
+			succeed: true,
+		},
+		"normal": {
+			in: tools.EtcdResponseWithError{
+				R: normalListResp,
+				E: nil,
+			},
+			m:       EverythingMatcher{},
+			out:     &api.PodList{Items: []api.Pod{*podA, *podB}},
+			succeed: true,
+		},
+		"normalFiltered": {
+			in: tools.EtcdResponseWithError{
+				R: normalListResp,
+				E: nil,
+			},
+			m:       SetMatcher{util.NewStringSet("foo")},
+			out:     &api.PodList{Items: []api.Pod{*podA}},
+			succeed: true,
+		},
+	}
+
+	for name, item := range table {
+		fakeClient, registry := NewTestGenericEtcdRegistry(t)
+		fakeClient.Data[registry.KeyRoot] = item.in
+		list, err := registry.List(api.NewContext(), item.m)
+		if e, a := item.succeed, err == nil; e != a {
+			t.Errorf("%v: expected %v, got %v", name, e, a)
+			continue
+		}
+
+		if e, a := item.out, list; !reflect.DeepEqual(e, a) {
+			t.Errorf("%v: Expected %#v, got %#v", name, e, a)
+		}
+	}
+}
+
+func TestEtcdCreate(t *testing.T) {
+	podA := &api.Pod{
+		TypeMeta:     api.TypeMeta{ID: "foo"},
+		DesiredState: api.PodState{Host: "machine"},
+	}
+	podB := &api.Pod{
+		TypeMeta:     api.TypeMeta{ID: "foo"},
+		DesiredState: api.PodState{Host: "machine2"},
+	}
+
+	nodeWithPodA := tools.EtcdResponseWithError{
+		R: &etcd.Response{
+			Node: &etcd.Node{
+				Value:         runtime.EncodeOrDie(testapi.Codec(), podA),
+				ModifiedIndex: 1,
+				CreatedIndex:  1,
+			},
+		},
+		E: nil,
+	}
+
+	emptyNode := tools.EtcdResponseWithError{
+		R: &etcd.Response{},
+		E: tools.EtcdErrorNotFound,
+	}
+
+	path := "/registry/pods/foo"
+	key := "foo"
+
+	table := map[string]struct {
+		existing tools.EtcdResponseWithError
+		expect   tools.EtcdResponseWithError
+		toCreate runtime.Object
+		errOK    func(error) bool
+	}{
+		"normal": {
+			existing: emptyNode,
+			expect:   nodeWithPodA,
+			toCreate: podA,
+			errOK:    func(err error) bool { return err == nil },
+		},
+		"preExisting": {
+			existing: nodeWithPodA,
+			expect:   nodeWithPodA,
+			toCreate: podB,
+			errOK:    errors.IsAlreadyExists,
+		},
+	}
+
+	for name, item := range table {
+		fakeClient, registry := NewTestGenericEtcdRegistry(t)
+		fakeClient.Data[path] = item.existing
+		err := registry.Create(api.NewContext(), key, item.toCreate)
+		if !item.errOK(err) {
+			t.Errorf("%v: unexpected error: %v", name, err)
+		}
+
+		if e, a := item.expect, fakeClient.Data[path]; !reflect.DeepEqual(e, a) {
+			t.Errorf("%v:\n%s", name, runtime.ObjectDiff(e, a))
+		}
+	}
+}
+
+func TestEtcdUpdate(t *testing.T) {
+	podA := &api.Pod{
+		TypeMeta:     api.TypeMeta{ID: "foo"},
+		DesiredState: api.PodState{Host: "machine"},
+	}
+	podB := &api.Pod{
+		TypeMeta:     api.TypeMeta{ID: "foo", ResourceVersion: "1"},
+		DesiredState: api.PodState{Host: "machine2"},
+	}
+
+	nodeWithPodA := tools.EtcdResponseWithError{
+		R: &etcd.Response{
+			Node: &etcd.Node{
+				Value:         runtime.EncodeOrDie(testapi.Codec(), podA),
+				ModifiedIndex: 1,
+				CreatedIndex:  1,
+			},
+		},
+		E: nil,
+	}
+
+	nodeWithPodB := tools.EtcdResponseWithError{
+		R: &etcd.Response{
+			Node: &etcd.Node{
+				Value:         runtime.EncodeOrDie(testapi.Codec(), podB),
+				ModifiedIndex: 1,
+				CreatedIndex:  1,
+			},
+		},
+		E: nil,
+	}
+
+	emptyNode := tools.EtcdResponseWithError{
+		R: &etcd.Response{},
+		E: tools.EtcdErrorNotFound,
+	}
+
+	path := "/registry/pods/foo"
+	key := "foo"
+
+	table := map[string]struct {
+		existing tools.EtcdResponseWithError
+		expect   tools.EtcdResponseWithError
+		toUpdate runtime.Object
+		errOK    func(error) bool
+	}{
+		"normal": {
+			existing: nodeWithPodA,
+			expect:   nodeWithPodB,
+			toUpdate: podB,
+			errOK:    func(err error) bool { return err == nil },
+		},
+		"notExisting": {
+			existing: emptyNode,
+			expect:   nodeWithPodA,
+			toUpdate: podA,
+			// TODO: Should updating a non-existing thing fail?
+			errOK: func(err error) bool { return err == nil },
+		},
+	}
+
+	for name, item := range table {
+		fakeClient, registry := NewTestGenericEtcdRegistry(t)
+		fakeClient.Data[path] = item.existing
+		err := registry.Update(api.NewContext(), key, item.toUpdate)
+		if !item.errOK(err) {
+			t.Errorf("%v: unexpected error: %v", name, err)
+		}
+
+		if e, a := item.expect, fakeClient.Data[path]; !reflect.DeepEqual(e, a) {
+			t.Errorf("%v:\n%s", name, runtime.ObjectDiff(e, a))
+		}
+	}
+}
+
+func TestEtcdGet(t *testing.T) {
+	podA := &api.Pod{
+		TypeMeta:     api.TypeMeta{ID: "foo", ResourceVersion: "1"},
+		DesiredState: api.PodState{Host: "machine"},
+	}
+
+	nodeWithPodA := tools.EtcdResponseWithError{
+		R: &etcd.Response{
+			Node: &etcd.Node{
+				Value:         runtime.EncodeOrDie(testapi.Codec(), podA),
+				ModifiedIndex: 1,
+				CreatedIndex:  1,
+			},
+		},
+		E: nil,
+	}
+
+	emptyNode := tools.EtcdResponseWithError{
+		R: &etcd.Response{},
+		E: tools.EtcdErrorNotFound,
+	}
+
+	path := "/registry/pods/foo"
+	key := "foo"
+
+	table := map[string]struct {
+		existing tools.EtcdResponseWithError
+		expect   runtime.Object
+		errOK    func(error) bool
+	}{
+		"normal": {
+			existing: nodeWithPodA,
+			expect:   podA,
+			errOK:    func(err error) bool { return err == nil },
+		},
+		"notExisting": {
+			existing: emptyNode,
+			expect:   nil,
+			errOK:    errors.IsNotFound,
+		},
+	}
+
+	for name, item := range table {
+		fakeClient, registry := NewTestGenericEtcdRegistry(t)
+		fakeClient.Data[path] = item.existing
+		got, err := registry.Get(api.NewContext(), key)
+		if !item.errOK(err) {
+			t.Errorf("%v: unexpected error: %v", name, err)
+		}
+
+		if e, a := item.expect, got; !reflect.DeepEqual(e, a) {
+			t.Errorf("%v:\n%s", name, runtime.ObjectDiff(e, a))
+		}
+	}
+}
+
+func TestEtcdDelete(t *testing.T) {
+	podA := &api.Pod{
+		TypeMeta:     api.TypeMeta{ID: "foo", ResourceVersion: "1"},
+		DesiredState: api.PodState{Host: "machine"},
+	}
+
+	nodeWithPodA := tools.EtcdResponseWithError{
+		R: &etcd.Response{
+			Node: &etcd.Node{
+				Value:         runtime.EncodeOrDie(testapi.Codec(), podA),
+				ModifiedIndex: 1,
+				CreatedIndex:  1,
+			},
+		},
+		E: nil,
+	}
+
+	emptyNode := tools.EtcdResponseWithError{
+		R: &etcd.Response{},
+		E: tools.EtcdErrorNotFound,
+	}
+
+	path := "/registry/pods/foo"
+	key := "foo"
+
+	table := map[string]struct {
+		existing tools.EtcdResponseWithError
+		expect   tools.EtcdResponseWithError
+		errOK    func(error) bool
+	}{
+		"normal": {
+			existing: nodeWithPodA,
+			expect:   emptyNode,
+			errOK:    func(err error) bool { return err == nil },
+		},
+		"notExisting": {
+			existing: emptyNode,
+			expect:   emptyNode,
+			errOK:    func(err error) bool { return err == nil },
+		},
+	}
+
+	for name, item := range table {
+		fakeClient, registry := NewTestGenericEtcdRegistry(t)
+		fakeClient.Data[path] = item.existing
+		err := registry.Delete(api.NewContext(), key)
+		if !item.errOK(err) {
+			t.Errorf("%v: unexpected error: %v", name, err)
+		}
+
+		if e, a := item.expect, fakeClient.Data[path]; !reflect.DeepEqual(e, a) {
+			t.Errorf("%v:\n%s", name, runtime.ObjectDiff(e, a))
+		}
+	}
+}
+
+func TestEtcdWatch(t *testing.T) {
+	podA := &api.Pod{
+		TypeMeta:     api.TypeMeta{ID: "foo", ResourceVersion: "1"},
+		DesiredState: api.PodState{Host: "machine"},
+	}
+	respWithPodA := &etcd.Response{
+		Node: &etcd.Node{
+			Value:         runtime.EncodeOrDie(testapi.Codec(), podA),
+			ModifiedIndex: 1,
+			CreatedIndex:  1,
+		},
+		Action: "create",
+	}
+
+	fakeClient, registry := NewTestGenericEtcdRegistry(t)
+	wi, err := registry.Watch(api.NewContext(), EverythingMatcher{}, 1)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	fakeClient.WaitForWatchCompletion()
+
+	go func() {
+		fakeClient.WatchResponse <- respWithPodA
+	}()
+
+	got, open := <-wi.ResultChan()
+	if !open {
+		t.Fatalf("unexpected channel close")
+	}
+
+	if e, a := podA, got.Object; !reflect.DeepEqual(e, a) {
+		t.Errorf("difference: %s", runtime.ObjectDiff(e, a))
+	}
+}

pkg/registry/generic/registry.go

@@ -0,0 +1,67 @@
+/*
+Copyright 2014 Google Inc. All rights reserved.
+
+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 generic
+
+import (
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
+)
+
+// AttrFunc returns label and field sets for List or Watch to compare against, or an error.
+type AttrFunc func(obj runtime.Object) (label, field labels.Set, err error)
+
+// SelectionPredicate implements a generic predicate that can be passed to
+// GenericRegistry's List or Watch methods. Implements the Matcher interface.
+type SelectionPredicate struct {
+	Label    labels.Selector
+	Field    labels.Selector
+	GetAttrs AttrFunc
+}
+
+// Matches returns true if the given object's labels and fields (as
+// returned by s.GetAttrs) match s.Label and s.Field. An error is
+// returned if s.GetAttrs fails.
+func (s *SelectionPredicate) Matches(obj runtime.Object) (bool, error) {
+	if s.Label.Empty() && s.Field.Empty() {
+		return true, nil
+	}
+	labels, fields, err := s.GetAttrs(obj)
+	if err != nil {
+		return false, err
+	}
+	return s.Label.Matches(labels) && s.Field.Matches(fields), nil
+}
+
+// Matcher can return true if an object matches the Matcher's selection
+// criteria.
+type Matcher interface {
+	Matches(obj runtime.Object) (bool, error)
+}
+
+// Registry knows how to store & list any runtime.Object. Can be used for
+// any object types which don't require special features from the storage
+// layer.
+type Registry interface {
+	List(api.Context, Matcher) (runtime.Object, error)
+	Create(ctx api.Context, id string, obj runtime.Object) error
+	Update(ctx api.Context, id string, obj runtime.Object) error
+	Get(ctx api.Context, id string) (runtime.Object, error)
+	Delete(ctx api.Context, id string) error
+	Watch(ctx api.Context, m Matcher, resourceVersion uint64) (watch.Interface, error)
+}

pkg/registry/generic/registry_test.go

@@ -0,0 +1,92 @@
+/*
+Copyright 2014 Google Inc. All rights reserved.
+
+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 generic
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
+)
+
+type Ignored struct{}
+
+func (*Ignored) IsAnAPIObject() {}
+
+func TestSelectionPredicate(t *testing.T) {
+	table := map[string]struct {
+		labelSelector, fieldSelector string
+		labels, fields               labels.Set
+		err                          error
+		shouldMatch                  bool
+	}{
+		"A": {
+			labelSelector: "name=foo",
+			fieldSelector: "uid=12345",
+			labels:        labels.Set{"name": "foo"},
+			fields:        labels.Set{"uid": "12345"},
+			shouldMatch:   true,
+		},
+		"B": {
+			labelSelector: "name=foo",
+			fieldSelector: "uid=12345",
+			labels:        labels.Set{"name": "foo"},
+			fields:        labels.Set{},
+			shouldMatch:   false,
+		},
+		"C": {
+			labelSelector: "name=foo",
+			fieldSelector: "uid=12345",
+			labels:        labels.Set{},
+			fields:        labels.Set{"uid": "12345"},
+			shouldMatch:   false,
+		},
+		"error": {
+			labelSelector: "name=foo",
+			fieldSelector: "uid=12345",
+			err:           errors.New("maybe this is a 'wrong object type' error"),
+			shouldMatch:   false,
+		},
+	}
+
+	for name, item := range table {
+		parsedLabel, err := labels.ParseSelector(item.labelSelector)
+		if err != nil {
+			panic(err)
+		}
+		parsedField, err := labels.ParseSelector(item.fieldSelector)
+		if err != nil {
+			panic(err)
+		}
+		sp := &SelectionPredicate{
+			Label: parsedLabel,
+			Field: parsedField,
+			GetAttrs: func(runtime.Object) (label, field labels.Set, err error) {
+				return item.labels, item.fields, item.err
+			},
+		}
+		got, err := sp.Matches(&Ignored{})
+		if e, a := item.err, err; e != a {
+			t.Errorf("%v: expected %v, got %v", name, e, a)
+			continue
+		}
+		if e, a := item.shouldMatch, got; e != a {
+			t.Errorf("%v: expected %v, got %v", name, e, a)
+		}
+	}
+}

pkg/registry/registrytest/generic.go

@@ -0,0 +1,87 @@
+/*
+Copyright 2014 Google Inc. All rights reserved.
+
+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 registrytest
+
+import (
+	"sync"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/registry/generic"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
+)
+
+// GenericRegistry knows how to store & list any runtime.Object. Events don't require
+// any non-generic features from the storage layer.
+type GenericRegistry struct {
+	Err        error
+	Object     runtime.Object
+	ObjectList runtime.Object
+	sync.Mutex
+
+	Mux *watch.Mux
+}
+
+func NewGeneric(list runtime.Object) *GenericRegistry {
+	return &GenericRegistry{
+		ObjectList: list,
+		Mux:        watch.NewMux(0),
+	}
+}
+
+func (r *GenericRegistry) List(ctx api.Context, m generic.Matcher) (runtime.Object, error) {
+	r.Lock()
+	defer r.Unlock()
+	if r.Err != nil {
+		return nil, r.Err
+	}
+	return generic.FilterList(r.ObjectList, m)
+}
+
+func (r *GenericRegistry) Watch(ctx api.Context, m generic.Matcher, resourceVersion uint64) (watch.Interface, error) {
+	// TODO: wire filter down into the mux; it needs access to current and previous state :(
+	return r.Mux.Watch(), nil
+}
+
+func (r *GenericRegistry) Get(ctx api.Context, id string) (runtime.Object, error) {
+	r.Lock()
+	defer r.Unlock()
+	return r.Object, r.Err
+}
+
+func (r *GenericRegistry) Create(ctx api.Context, id string, obj runtime.Object) error {
+	r.Lock()
+	defer r.Unlock()
+	r.Object = obj
+	r.Mux.Action(watch.Added, obj)
+	return r.Err
+}
+
+func (r *GenericRegistry) Update(ctx api.Context, id string, obj runtime.Object) error {
+	r.Lock()
+	defer r.Unlock()
+	r.Object = obj
+	r.Mux.Action(watch.Modified, obj)
+	return r.Err
+}
+
+func (r *GenericRegistry) Delete(ctx api.Context, id string) error {
+	r.Lock()
+	defer r.Unlock()
+	r.Mux.Action(watch.Deleted, r.Object)
+	return r.Err
+}

pkg/tools/fake_etcd_client.go

@@ -167,6 +167,7 @@ func (f *FakeEtcdClient) setLocked(key, value string, ttl uint64) (*etcd.Respons
 				Value:         value,
 				CreatedIndex:  i,
 				ModifiedIndex: i,
+				TTL:           int64(ttl),
 			},
 		},
 	}