Merge pull request #23490 from wojtek-t/remove_set_from_storage_interface

pkg/master/master_test.go

@@ -569,7 +569,7 @@ func testInstallThirdPartyAPIListVersion(t *testing.T, version string) {
 			defer etcdserver.Terminate(t)
 
 			if test.items != nil {
-				err := storeThirdPartyList(master.thirdPartyStorage, "/ThirdPartyResourceData/company.com/foos/default", test.items)
+				err := createThirdPartyList(master.thirdPartyStorage, "/ThirdPartyResourceData/company.com/foos/default", test.items)
 				if !assert.NoError(err) {
 					return
 				}
@@ -637,17 +637,17 @@ func encodeToThirdParty(name string, obj interface{}) (runtime.Object, error) {
 	return &thirdPartyData, nil
 }
 
-func storeThirdPartyObject(s storage.Interface, path, name string, obj interface{}) error {
+func createThirdPartyObject(s storage.Interface, path, name string, obj interface{}) error {
 	data, err := encodeToThirdParty(name, obj)
 	if err != nil {
 		return err
 	}
-	return s.Set(context.TODO(), etcdtest.AddPrefix(path), data, nil, 0)
+	return s.Create(context.TODO(), etcdtest.AddPrefix(path), data, nil, 0)
 }
 
-func storeThirdPartyList(s storage.Interface, path string, list []Foo) error {
+func createThirdPartyList(s storage.Interface, path string, list []Foo) error {
 	for _, obj := range list {
-		if err := storeThirdPartyObject(s, path+"/"+obj.Name, obj.Name, obj); err != nil {
+		if err := createThirdPartyObject(s, path+"/"+obj.Name, obj.Name, obj); err != nil {
 			return err
 		}
 	}
@@ -690,7 +690,7 @@ func testInstallThirdPartyAPIGetVersion(t *testing.T, version string) {
 		SomeField:  "test field",
 		OtherField: 10,
 	}
-	if !assert.NoError(storeThirdPartyObject(master.thirdPartyStorage, "/ThirdPartyResourceData/company.com/foos/default/test", "test", expectedObj)) {
+	if !assert.NoError(createThirdPartyObject(master.thirdPartyStorage, "/ThirdPartyResourceData/company.com/foos/default/test", "test", expectedObj)) {
 		t.FailNow()
 		return
 	}
@@ -803,7 +803,7 @@ func testInstallThirdPartyAPIDeleteVersion(t *testing.T, version string) {
 		SomeField:  "test field",
 		OtherField: 10,
 	}
-	if !assert.NoError(storeThirdPartyObject(master.thirdPartyStorage, "/ThirdPartyResourceData/company.com/foos/default/test", "test", expectedObj)) {
+	if !assert.NoError(createThirdPartyObject(master.thirdPartyStorage, "/ThirdPartyResourceData/company.com/foos/default/test", "test", expectedObj)) {
 		t.FailNow()
 		return
 	}
@@ -913,13 +913,13 @@ func testInstallThirdPartyResourceRemove(t *testing.T, version string) {
 		SomeField:  "test field",
 		OtherField: 10,
 	}
-	if !assert.NoError(storeThirdPartyObject(master.thirdPartyStorage, "/ThirdPartyResourceData/company.com/foos/default/test", "test", expectedObj)) {
+	if !assert.NoError(createThirdPartyObject(master.thirdPartyStorage, "/ThirdPartyResourceData/company.com/foos/default/test", "test", expectedObj)) {
 		t.FailNow()
 		return
 	}
 	secondObj := expectedObj
 	secondObj.Name = "bar"
-	if !assert.NoError(storeThirdPartyObject(master.thirdPartyStorage, "/ThirdPartyResourceData/company.com/foos/default/bar", "bar", secondObj)) {
+	if !assert.NoError(createThirdPartyObject(master.thirdPartyStorage, "/ThirdPartyResourceData/company.com/foos/default/bar", "bar", secondObj)) {
 		t.FailNow()
 		return
 	}

pkg/registry/deployment/etcd/etcd_test.go

@@ -185,7 +185,7 @@ func TestScaleGet(t *testing.T) {
 	var deployment extensions.Deployment
 	ctx := api.WithNamespace(api.NewContext(), namespace)
 	key := etcdtest.AddPrefix("/deployments/" + namespace + "/" + name)
-	if err := storage.Deployment.Storage.Set(ctx, key, &validDeployment, &deployment, 0); err != nil {
+	if err := storage.Deployment.Storage.Create(ctx, key, &validDeployment, &deployment, 0); err != nil {
 		t.Fatalf("error setting new deployment (key: %s) %v: %v", key, validDeployment, err)
 	}
 
@@ -222,7 +222,7 @@ func TestScaleUpdate(t *testing.T) {
 	var deployment extensions.Deployment
 	ctx := api.WithNamespace(api.NewContext(), namespace)
 	key := etcdtest.AddPrefix("/deployments/" + namespace + "/" + name)
-	if err := storage.Deployment.Storage.Set(ctx, key, &validDeployment, &deployment, 0); err != nil {
+	if err := storage.Deployment.Storage.Create(ctx, key, &validDeployment, &deployment, 0); err != nil {
 		t.Fatalf("error setting new deployment (key: %s) %v: %v", key, validDeployment, err)
 	}
 	replicas := 12
@@ -259,7 +259,7 @@ func TestStatusUpdate(t *testing.T) {
 
 	ctx := api.WithNamespace(api.NewContext(), namespace)
 	key := etcdtest.AddPrefix("/deployments/" + namespace + "/" + name)
-	if err := storage.Deployment.Storage.Set(ctx, key, &validDeployment, nil, 0); err != nil {
+	if err := storage.Deployment.Storage.Create(ctx, key, &validDeployment, nil, 0); err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
 	update := extensions.Deployment{

pkg/registry/experimental/controller/etcd/etcd_test.go

@@ -86,7 +86,7 @@ func TestGet(t *testing.T) {
 
 	ctx := api.WithNamespace(api.NewContext(), "test")
 	key := etcdtest.AddPrefix("/controllers/test/foo")
-	if err := si.Set(ctx, key, &validController, nil, 0); err != nil {
+	if err := si.Create(ctx, key, &validController, nil, 0); err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
 	obj, err := storage.Get(ctx, "foo")
@@ -105,7 +105,7 @@ func TestUpdate(t *testing.T) {
 
 	ctx := api.WithNamespace(api.NewContext(), "test")
 	key := etcdtest.AddPrefix("/controllers/test/foo")
-	if err := si.Set(ctx, key, &validController, nil, 0); err != nil {
+	if err := si.Create(ctx, key, &validController, nil, 0); err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
 	replicas := 12

pkg/registry/namespace/etcd/etcd_test.go

@@ -147,7 +147,7 @@ func TestDeleteNamespaceWithIncompleteFinalizers(t *testing.T) {
 		},
 		Status: api.NamespaceStatus{Phase: api.NamespaceActive},
 	}
-	if err := storage.Storage.Set(ctx, key, namespace, nil, 0); err != nil {
+	if err := storage.Storage.Create(ctx, key, namespace, nil, 0); err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
 	if _, err := storage.Delete(ctx, "foo", nil); err == nil {
@@ -171,7 +171,7 @@ func TestDeleteNamespaceWithCompleteFinalizers(t *testing.T) {
 		},
 		Status: api.NamespaceStatus{Phase: api.NamespaceActive},
 	}
-	if err := storage.Storage.Set(ctx, key, namespace, nil, 0); err != nil {
+	if err := storage.Storage.Create(ctx, key, namespace, nil, 0); err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
 	if _, err := storage.Delete(ctx, "foo", nil); err != nil {

pkg/registry/persistentvolume/etcd/etcd_test.go

@@ -153,7 +153,7 @@ func TestUpdateStatus(t *testing.T) {
 	key, _ := storage.KeyFunc(ctx, "foo")
 	key = etcdtest.AddPrefix(key)
 	pvStart := validNewPersistentVolume("foo")
-	err := storage.Storage.Set(ctx, key, pvStart, nil, 0)
+	err := storage.Storage.Create(ctx, key, pvStart, nil, 0)
 	if err != nil {
 		t.Errorf("unexpected error: %v", err)
 	}

pkg/registry/persistentvolumeclaim/etcd/etcd_test.go

@@ -148,7 +148,7 @@ func TestUpdateStatus(t *testing.T) {
 	key, _ := storage.KeyFunc(ctx, "foo")
 	key = etcdtest.AddPrefix(key)
 	pvcStart := validNewPersistentVolumeClaim("foo", api.NamespaceDefault)
-	err := storage.Storage.Set(ctx, key, pvcStart, nil, 0)
+	err := storage.Storage.Create(ctx, key, pvcStart, nil, 0)
 
 	pvc := &api.PersistentVolumeClaim{
 		ObjectMeta: api.ObjectMeta{

pkg/registry/registrytest/etcd.go

@@ -82,7 +82,7 @@ func (t *Tester) ReturnDeletedObject() *Tester {
 func (t *Tester) TestCreate(valid runtime.Object, invalid ...runtime.Object) {
 	t.tester.TestCreate(
 		valid,
-		t.setObject,
+		t.createObject,
 		t.getObject,
 		invalid...,
 	)
@@ -95,7 +95,7 @@ func (t *Tester) TestUpdate(valid runtime.Object, validUpdateFunc UpdateFunc, in
 	}
 	t.tester.TestUpdate(
 		valid,
-		t.setObject,
+		t.createObject,
 		t.getObject,
 		resttest.UpdateFunc(validUpdateFunc),
 		invalidFuncs...,
@@ -105,7 +105,7 @@ func (t *Tester) TestUpdate(valid runtime.Object, validUpdateFunc UpdateFunc, in
 func (t *Tester) TestDelete(valid runtime.Object) {
 	t.tester.TestDelete(
 		valid,
-		t.setObject,
+		t.createObject,
 		t.getObject,
 		errors.IsNotFound,
 	)
@@ -114,7 +114,7 @@ func (t *Tester) TestDelete(valid runtime.Object) {
 func (t *Tester) TestDeleteGraceful(valid runtime.Object, expectedGrace int64) {
 	t.tester.TestDeleteGraceful(
 		valid,
-		t.setObject,
+		t.createObject,
 		t.getObject,
 		expectedGrace,
 	)
@@ -181,7 +181,7 @@ func (t *Tester) getObject(ctx api.Context, obj runtime.Object) (runtime.Object,
 	return result, nil
 }
 
-func (t *Tester) setObject(ctx api.Context, obj runtime.Object) error {
+func (t *Tester) createObject(ctx api.Context, obj runtime.Object) error {
 	accessor, err := meta.Accessor(obj)
 	if err != nil {
 		return err
@@ -190,7 +190,7 @@ func (t *Tester) setObject(ctx api.Context, obj runtime.Object) error {
 	if err != nil {
 		return err
 	}
-	return t.storage.Storage.Set(ctx, key, obj, nil, 0)
+	return t.storage.Storage.Create(ctx, key, obj, nil, 0)
 }
 
 func (t *Tester) setObjectsForList(objects []runtime.Object) []runtime.Object {
@@ -208,7 +208,7 @@ func (t *Tester) emitObject(obj runtime.Object, action string) error {
 
 	switch action {
 	case etcdstorage.EtcdCreate:
-		err = t.setObject(ctx, obj)
+		err = t.createObject(ctx, obj)
 	case etcdstorage.EtcdDelete:
 		accessor, err := meta.Accessor(obj)
 		if err != nil {

pkg/registry/replicaset/etcd/etcd_test.go

@@ -246,7 +246,7 @@ func TestScaleGet(t *testing.T) {
 	var rs extensions.ReplicaSet
 	ctx := api.WithNamespace(api.NewContext(), api.NamespaceDefault)
 	key := etcdtest.AddPrefix("/replicasets/" + api.NamespaceDefault + "/" + name)
-	if err := storage.ReplicaSet.Storage.Set(ctx, key, &validReplicaSet, &rs, 0); err != nil {
+	if err := storage.ReplicaSet.Storage.Create(ctx, key, &validReplicaSet, &rs, 0); err != nil {
 		t.Fatalf("error setting new replica set (key: %s) %v: %v", key, validReplicaSet, err)
 	}
 
@@ -285,7 +285,7 @@ func TestScaleUpdate(t *testing.T) {
 	var rs extensions.ReplicaSet
 	ctx := api.WithNamespace(api.NewContext(), api.NamespaceDefault)
 	key := etcdtest.AddPrefix("/replicasets/" + api.NamespaceDefault + "/" + name)
-	if err := storage.ReplicaSet.Storage.Set(ctx, key, &validReplicaSet, &rs, 0); err != nil {
+	if err := storage.ReplicaSet.Storage.Create(ctx, key, &validReplicaSet, &rs, 0); err != nil {
 		t.Fatalf("error setting new replica set (key: %s) %v: %v", key, validReplicaSet, err)
 	}
 	replicas := 12
@@ -326,7 +326,7 @@ func TestStatusUpdate(t *testing.T) {
 
 	ctx := api.WithNamespace(api.NewContext(), api.NamespaceDefault)
 	key := etcdtest.AddPrefix("/replicasets/" + api.NamespaceDefault + "/foo")
-	if err := storage.ReplicaSet.Storage.Set(ctx, key, &validReplicaSet, nil, 0); err != nil {
+	if err := storage.ReplicaSet.Storage.Create(ctx, key, &validReplicaSet, nil, 0); err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
 	update := extensions.ReplicaSet{

pkg/registry/resourcequota/etcd/etcd_test.go

@@ -147,7 +147,7 @@ func TestUpdateStatus(t *testing.T) {
 	key, _ := storage.KeyFunc(ctx, "foo")
 	key = etcdtest.AddPrefix(key)
 	resourcequotaStart := validNewResourceQuota()
-	err := storage.Storage.Set(ctx, key, resourcequotaStart, nil, 0)
+	err := storage.Storage.Create(ctx, key, resourcequotaStart, nil, 0)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}

pkg/registry/service/allocator/etcd/etcd_test.go

@@ -58,7 +58,7 @@ func TestEmpty(t *testing.T) {
 func TestStore(t *testing.T) {
 	storage, server, backing := newStorage(t)
 	defer server.Terminate(t)
-	if err := storage.storage.Set(context.TODO(), key(), validNewRangeAllocation(), nil, 0); err != nil {
+	if err := storage.storage.Create(context.TODO(), key(), validNewRangeAllocation(), nil, 0); err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
 

pkg/registry/service/ipallocator/etcd/etcd_test.go

@@ -82,7 +82,7 @@ func TestErrors(t *testing.T) {
 func TestStore(t *testing.T) {
 	server, storage, backing, si := newStorage(t)
 	defer server.Terminate(t)
-	if err := si.Set(context.TODO(), key(), validNewRangeAllocation(), nil, 0); err != nil {
+	if err := si.Create(context.TODO(), key(), validNewRangeAllocation(), nil, 0); err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
 

pkg/storage/cacher.go

@@ -235,11 +235,6 @@ func (c *Cacher) Create(ctx context.Context, key string, obj, out runtime.Object
 }
 
 // Implements storage.Interface.
-func (c *Cacher) Set(ctx context.Context, key string, obj, out runtime.Object, ttl uint64) error {
-	return c.storage.Set(ctx, key, obj, out, ttl)
-}
-
-// Implements storage.Interface.
 func (c *Cacher) Delete(ctx context.Context, key string, out runtime.Object, preconditions *Preconditions) error {
 	return c.storage.Delete(ctx, key, out, preconditions)
 }

pkg/storage/cacher_test.go

@@ -70,20 +70,22 @@ func makeTestPod(name string) *api.Pod {
 }
 
 func updatePod(t *testing.T, s storage.Interface, obj, old *api.Pod) *api.Pod {
-	key := etcdtest.AddPrefix("pods/ns/" + obj.Name)
-	result := &api.Pod{}
-	if old == nil {
-		if err := s.Create(context.TODO(), key, obj, result, 0); err != nil {
-			t.Errorf("unexpected error: %v", err)
-		}
-	} else {
-		// To force "update" behavior of Set() we need to set ResourceVersion of
-		// previous version of object.
-		obj.ResourceVersion = old.ResourceVersion
-		if err := s.Set(context.TODO(), key, obj, result, 0); err != nil {
+	updateFn := func(input runtime.Object, res storage.ResponseMeta) (runtime.Object, *uint64, error) {
+		newObj, err := api.Scheme.DeepCopy(obj)
+		if err != nil {
 			t.Errorf("unexpected error: %v", err)
+			return nil, nil, err
 		}
-		obj.ResourceVersion = ""
+		return newObj.(*api.Pod), nil, nil
+	}
+	key := etcdtest.AddPrefix("pods/ns/" + obj.Name)
+	if err := s.GuaranteedUpdate(context.TODO(), key, &api.Pod{}, old == nil, nil, updateFn); err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+	obj.ResourceVersion = ""
+	result := &api.Pod{}
+	if err := s.Get(context.TODO(), key, result, false); err != nil {
+		t.Errorf("unexpected error: %v", err)
 	}
 	return result
 }

pkg/storage/etcd/etcd_helper.go

@@ -228,69 +228,6 @@ func (h *etcdHelper) Create(ctx context.Context, key string, obj, out runtime.Ob
 	return err
 }
 
-// Implements storage.Interface.
-func (h *etcdHelper) Set(ctx context.Context, key string, obj, out runtime.Object, ttl uint64) error {
-	if ctx == nil {
-		glog.Errorf("Context is nil")
-	}
-
-	version := uint64(0)
-	var err error
-	if version, err = h.versioner.ObjectResourceVersion(obj); err != nil {
-		return errors.New("couldn't get resourceVersion from object")
-	}
-	if version != 0 {
-		// We cannot store object with resourceVersion in etcd, we need to clear it here.
-		if err := h.versioner.UpdateObject(obj, nil, 0); err != nil {
-			return errors.New("resourceVersion cannot be set on objects store in etcd")
-		}
-	}
-
-	var response *etcd.Response
-	data, err := runtime.Encode(h.codec, obj)
-	if err != nil {
-		return err
-	}
-	key = h.prefixEtcdKey(key)
-
-	create := true
-	if version != 0 {
-		create = false
-		startTime := time.Now()
-		opts := etcd.SetOptions{
-			TTL:       time.Duration(ttl) * time.Second,
-			PrevIndex: version,
-		}
-		response, err = h.etcdKeysAPI.Set(ctx, key, string(data), &opts)
-		metrics.RecordEtcdRequestLatency("compareAndSwap", getTypeName(obj), startTime)
-		if err != nil {
-			return toStorageErr(err, key, int64(version))
-		}
-	}
-	if create {
-		// Create will fail if a key already exists.
-		startTime := time.Now()
-		opts := etcd.SetOptions{
-			TTL:       time.Duration(ttl) * time.Second,
-			PrevExist: etcd.PrevNoExist,
-		}
-		response, err = h.etcdKeysAPI.Set(ctx, key, string(data), &opts)
-		if err != nil {
-			return toStorageErr(err, key, 0)
-		}
-		metrics.RecordEtcdRequestLatency("create", getTypeName(obj), startTime)
-	}
-
-	if out != nil {
-		if _, err := conversion.EnforcePtr(out); err != nil {
-			panic("unable to convert output object to pointer")
-		}
-		_, _, err = h.extractObj(response, err, out, false, false)
-	}
-
-	return err
-}
-
 func checkPreconditions(preconditions *storage.Preconditions, out runtime.Object) error {
 	if preconditions == nil {
 		return nil

pkg/storage/etcd/etcd_helper_test.go

@@ -229,7 +229,7 @@ func TestGet(t *testing.T) {
 		Spec:       apitesting.DeepEqualSafePodSpec(),
 	}
 	var got api.Pod
-	if err := helper.Set(context.TODO(), key, &expect, &got, 0); err != nil {
+	if err := helper.Create(context.TODO(), key, &expect, &got, 0); err != nil {
 		t.Errorf("Unexpected error %#v", err)
 	}
 	expect = got
@@ -293,76 +293,6 @@ func TestCreateNilOutParam(t *testing.T) {
 	}
 }
 
-func TestSet(t *testing.T) {
-	obj := &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}}
-	server := etcdtesting.NewEtcdTestClientServer(t)
-	defer server.Terminate(t)
-	helper := newEtcdHelper(server.Client, testapi.Default.Codec(), etcdtest.PathPrefix())
-	returnedObj := &api.Pod{}
-	err := helper.Set(context.TODO(), "/some/key", obj, returnedObj, 5)
-	if err != nil {
-		t.Errorf("Unexpected error %#v", err)
-	}
-
-	if obj.ObjectMeta.Name == returnedObj.ObjectMeta.Name {
-		// Set worked, now override the values.
-		obj = returnedObj
-	}
-
-	err = helper.Get(context.TODO(), "/some/key", returnedObj, false)
-	if err != nil {
-		t.Errorf("Unexpected error %#v", err)
-	}
-	if !reflect.DeepEqual(obj, returnedObj) {
-		t.Errorf("Wanted %#v, got %#v", obj, returnedObj)
-	}
-}
-
-func TestSetFailCAS(t *testing.T) {
-	obj := &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo", ResourceVersion: "1"}}
-	server := etcdtesting.NewEtcdTestClientServer(t)
-	defer server.Terminate(t)
-	helper := newEtcdHelper(server.Client, testapi.Default.Codec(), etcdtest.PathPrefix())
-	err := helper.Set(context.TODO(), "/some/key", obj, nil, 5)
-	if err == nil {
-		t.Errorf("Expecting error.")
-	}
-}
-
-func TestSetWithVersion(t *testing.T) {
-	obj := &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}}
-	server := etcdtesting.NewEtcdTestClientServer(t)
-	defer server.Terminate(t)
-	helper := newEtcdHelper(server.Client, testapi.Default.Codec(), etcdtest.PathPrefix())
-
-	returnedObj := &api.Pod{}
-	err := helper.Set(context.TODO(), "/some/key", obj, returnedObj, 7)
-	if err != nil {
-		t.Fatalf("Unexpected error %#v", err)
-	}
-	// resource revision is now set, try to set again with new value to test CAS
-	obj = returnedObj
-	obj.Name = "bar"
-	err = helper.Set(context.TODO(), "/some/key", obj, returnedObj, 7)
-	if err != nil {
-		t.Fatalf("Unexpected error %#v", err)
-	}
-	if returnedObj.Name != "bar" {
-		t.Fatalf("Unexpected error %#v", returnedObj)
-	}
-}
-
-func TestSetNilOutParam(t *testing.T) {
-	obj := &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}}
-	server := etcdtesting.NewEtcdTestClientServer(t)
-	defer server.Terminate(t)
-	helper := newEtcdHelper(server.Client, testapi.Default.Codec(), etcdtest.PathPrefix())
-	err := helper.Set(context.TODO(), "/some/key", obj, nil, 3)
-	if err != nil {
-		t.Errorf("Unexpected error %#v", err)
-	}
-}
-
 func TestGuaranteedUpdate(t *testing.T) {
 	_, codec := testScheme(t)
 	server := etcdtesting.NewEtcdTestClientServer(t)

pkg/storage/etcd/etcd_watcher_test.go

@@ -236,7 +236,7 @@ func TestWatch(t *testing.T) {
 	// Test normal case
 	pod := &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}}
 	returnObj := &api.Pod{}
-	err = h.Set(context.TODO(), key, pod, returnObj, 0)
+	err = h.Create(context.TODO(), key, pod, returnObj, 0)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -292,7 +292,7 @@ func TestWatchEtcdState(t *testing.T) {
 		Subsets:    emptySubsets(),
 	}
 
-	err = h.Set(context.TODO(), key, endpoint, endpoint, 0)
+	err = h.Create(context.TODO(), key, endpoint, endpoint, 0)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -304,9 +304,18 @@ func TestWatchEtcdState(t *testing.T) {
 
 	subset := makeSubsets("127.0.0.1", 9000)
 	endpoint.Subsets = subset
+	endpoint.ResourceVersion = ""
 
 	// CAS the previous value
-	err = h.Set(context.TODO(), key, endpoint, endpoint, 0)
+	updateFn := func(input runtime.Object, res storage.ResponseMeta) (runtime.Object, *uint64, error) {
+		newObj, err := api.Scheme.DeepCopy(endpoint)
+		if err != nil {
+			t.Errorf("unexpected error: %v", err)
+			return nil, nil, err
+		}
+		return newObj.(*api.Endpoints), nil, nil
+	}
+	err = h.GuaranteedUpdate(context.TODO(), key, &api.Endpoints{}, false, nil, updateFn)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -317,7 +326,7 @@ func TestWatchEtcdState(t *testing.T) {
 	}
 
 	if e, a := endpoint, event.Object; !api.Semantic.DeepDerivative(e, a) {
-		t.Errorf("Unexpected error: expected %v, got %v", e, a)
+		t.Errorf("Unexpected error: expected %#v, got %#v", e, a)
 	}
 }
 
@@ -332,14 +341,19 @@ func TestWatchFromZeroIndex(t *testing.T) {
 	h := newEtcdHelper(server.Client, codec, etcdtest.PathPrefix())
 
 	// set before the watch and verify events
-	err := h.Set(context.TODO(), key, pod, pod, 0)
+	err := h.Create(context.TODO(), key, pod, pod, 0)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
+	pod.ResourceVersion = ""
 
 	// check for concatenation on watch event with CAS
-	pod.Name = "bar"
-	err = h.Set(context.TODO(), key, pod, pod, 0)
+	updateFn := func(input runtime.Object, res storage.ResponseMeta) (runtime.Object, *uint64, error) {
+		pod := input.(*api.Pod)
+		pod.Name = "bar"
+		return pod, nil, nil
+	}
+	err = h.GuaranteedUpdate(context.TODO(), key, &api.Pod{}, false, nil, updateFn)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -356,7 +370,13 @@ func TestWatchFromZeroIndex(t *testing.T) {
 		t.Errorf("Unexpected event %#v", event)
 	}
 
-	err = h.Set(context.TODO(), key, pod, pod, 0)
+	pod.Name = "baz"
+	updateFn = func(input runtime.Object, res storage.ResponseMeta) (runtime.Object, *uint64, error) {
+		pod := input.(*api.Pod)
+		pod.Name = "baz"
+		return pod, nil, nil
+	}
+	err = h.GuaranteedUpdate(context.TODO(), key, &api.Pod{}, false, nil, updateFn)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -367,7 +387,7 @@ func TestWatchFromZeroIndex(t *testing.T) {
 	}
 
 	if e, a := pod, event.Object; !api.Semantic.DeepDerivative(e, a) {
-		t.Errorf("Unexpected error: expected %v, got %v", e, a)
+		t.Errorf("Unexpected error: expected %#v, got %#v", e, a)
 	}
 }
 

pkg/storage/interfaces.go

@@ -98,11 +98,6 @@ type Interface interface {
 	// set to the read value from database.
 	Create(ctx context.Context, key string, obj, out runtime.Object, ttl uint64) error
 
-	// Set marshals obj via json and stores in database under key. Will do an atomic update
-	// if obj's ResourceVersion field is set. 'ttl' is time-to-live in seconds (0 means forever).
-	// If no error is returned and out is not nil, out will be set to the read value from database.
-	Set(ctx context.Context, key string, obj, out runtime.Object, ttl uint64) error
-
 	// Delete removes the specified key and returns the value that existed at that spot.
 	// If key didn't exist, it will return NotFound storage error.
 	Delete(ctx context.Context, key string, out runtime.Object, preconditions *Preconditions) error

pkg/storage/testing/utils.go

@@ -27,7 +27,7 @@ import (
 
 // CreateObj will create a single object using the storage interface
 func CreateObj(helper storage.Interface, name string, obj, out runtime.Object, ttl uint64) error {
-	return helper.Set(context.TODO(), name, obj, out, ttl)
+	return helper.Create(context.TODO(), name, obj, out, ttl)
 }
 
 //CreateObjList will create a list from the array of objects

test/integration/etcd_tools_test.go

@@ -35,14 +35,14 @@ import (
 	"golang.org/x/net/context"
 )
 
-func TestSet(t *testing.T) {
+func TestCreate(t *testing.T) {
 	client := framework.NewEtcdClient()
 	keysAPI := etcd.NewKeysAPI(client)
 	etcdStorage := etcdstorage.NewEtcdStorage(client, testapi.Default.Codec(), "", false)
 	ctx := context.TODO()
 	framework.WithEtcdKey(func(key string) {
 		testObject := api.ServiceAccount{ObjectMeta: api.ObjectMeta{Name: "foo"}}
-		if err := etcdStorage.Set(ctx, key, &testObject, nil, 0); err != nil {
+		if err := etcdStorage.Create(ctx, key, &testObject, nil, 0); err != nil {
 			t.Fatalf("unexpected error: %v", err)
 		}
 		resp, err := keysAPI.Get(ctx, key, nil)
@@ -94,7 +94,7 @@ func TestWriteTTL(t *testing.T) {
 	ctx := context.TODO()
 	framework.WithEtcdKey(func(key string) {
 		testObject := api.ServiceAccount{ObjectMeta: api.ObjectMeta{Name: "foo"}}
-		if err := etcdStorage.Set(ctx, key, &testObject, nil, 0); err != nil {
+		if err := etcdStorage.Create(ctx, key, &testObject, nil, 0); err != nil {
 			t.Fatalf("unexpected error: %v", err)
 		}
 		result := &api.ServiceAccount{}