Merge pull request #4157 from saad-ali/updateEventApi

Add "Update Event" to Kubernetes API

pkg/api/errors/etcd/etcd.go

@@ -43,7 +43,7 @@ func InterpretCreateError(err error, kind, name string) error {
 	}
 }
 
-// InterpretUpdateError converts a generic etcd error on a create
+// InterpretUpdateError converts a generic etcd error on a update
 // operation into the appropriate API error.
 func InterpretUpdateError(err error, kind, name string) error {
 	switch {
@@ -54,7 +54,7 @@ func InterpretUpdateError(err error, kind, name string) error {
 	}
 }
 
-// InterpretDeleteError converts a generic etcd error on a create
+// InterpretDeleteError converts a generic etcd error on a delete
 // operation into the appropriate API error.
 func InterpretDeleteError(err error, kind, name string) error {
 	switch {

pkg/registry/etcd/etcd.go

@@ -251,7 +251,7 @@ func (r *Registry) UpdatePod(ctx api.Context, pod *api.Pod) error {
 	}
 	// There's no race with the scheduler, because either this write will fail because the host
 	// has been updated, or the host update will fail because this pod has been updated.
-	err = r.EtcdHelper.SetObj(podKey, pod)
+	err = r.EtcdHelper.SetObj(podKey, pod, 0 /* ttl */)
 	if err != nil {
 		return err
 	}
@@ -404,7 +404,7 @@ func (r *Registry) UpdateController(ctx api.Context, controller *api.Replication
 	if err != nil {
 		return err
 	}
-	err = r.SetObj(key, controller)
+	err = r.SetObj(key, controller, 0 /* ttl */)
 	return etcderr.InterpretUpdateError(err, "replicationController", controller.Name)
 }
 
@@ -512,7 +512,7 @@ func (r *Registry) UpdateService(ctx api.Context, svc *api.Service) error {
 	if err != nil {
 		return err
 	}
-	err = r.SetObj(key, svc)
+	err = r.SetObj(key, svc, 0 /* ttl */)
 	return etcderr.InterpretUpdateError(err, "service", svc.Name)
 }
 
@@ -605,7 +605,7 @@ func (r *Registry) CreateMinion(ctx api.Context, minion *api.Node) error {
 
 func (r *Registry) UpdateMinion(ctx api.Context, minion *api.Node) error {
 	// TODO: Add some validations.
-	err := r.SetObj(makeNodeKey(minion.Name), minion)
+	err := r.SetObj(makeNodeKey(minion.Name), minion, 0 /* ttl */)
 	return etcderr.InterpretUpdateError(err, "minion", minion.Name)
 }
 

pkg/registry/event/registry.go

@@ -41,6 +41,17 @@ func (r registry) Create(ctx api.Context, id string, obj runtime.Object) error {
 	return etcderr.InterpretCreateError(err, r.Etcd.EndpointName, id)
 }
 
+// Update replaces an existing instance of the object, and sets a ttl so that the event
+// doesn't stay in the system forever.
+func (r registry) Update(ctx api.Context, id string, obj runtime.Object) error {
+	key, err := r.Etcd.KeyFunc(ctx, id)
+	if err != nil {
+		return err
+	}
+	err = r.Etcd.Helper.SetObj(key, obj, r.ttl)
+	return etcderr.InterpretUpdateError(err, r.Etcd.EndpointName, id)
+}
+
 // NewEtcdRegistry returns a registry which will store Events in the given
 // EtcdHelper. ttl is the time that Events will be retained by the system.
 func NewEtcdRegistry(h tools.EtcdHelper, ttl uint64) generic.Registry {

pkg/registry/event/registry_test.go

@@ -108,3 +108,105 @@ func TestEventCreate(t *testing.T) {
 		}
 	}
 }
+
+func TestEventUpdate(t *testing.T) {
+	eventA := &api.Event{
+		ObjectMeta: api.ObjectMeta{Name: "foo", Namespace: api.NamespaceDefault},
+		Reason:     "forTesting",
+	}
+	eventB := &api.Event{
+		ObjectMeta: api.ObjectMeta{Name: "bar", Namespace: api.NamespaceDefault},
+		Reason:     "for testing again",
+	}
+	eventC := &api.Event{
+		ObjectMeta: api.ObjectMeta{Name: "pan", Namespace: api.NamespaceDefault, ResourceVersion: "1"},
+		Reason:     "for testing again something else",
+	}
+
+	nodeWithEventA := tools.EtcdResponseWithError{
+		R: &etcd.Response{
+			Node: &etcd.Node{
+				Value:         runtime.EncodeOrDie(testapi.Codec(), eventA),
+				ModifiedIndex: 1,
+				CreatedIndex:  1,
+				TTL:           int64(testTTL),
+			},
+		},
+		E: nil,
+	}
+
+	nodeWithEventB := tools.EtcdResponseWithError{
+		R: &etcd.Response{
+			Node: &etcd.Node{
+				Value:         runtime.EncodeOrDie(testapi.Codec(), eventB),
+				ModifiedIndex: 1,
+				CreatedIndex:  1,
+				TTL:           int64(testTTL),
+			},
+		},
+		E: nil,
+	}
+
+	nodeWithEventC := tools.EtcdResponseWithError{
+		R: &etcd.Response{
+			Node: &etcd.Node{
+				Value:         runtime.EncodeOrDie(testapi.Codec(), eventC),
+				ModifiedIndex: 1,
+				CreatedIndex:  1,
+				TTL:           int64(testTTL),
+			},
+		},
+		E: nil,
+	}
+
+	emptyNode := tools.EtcdResponseWithError{
+		R: &etcd.Response{},
+		E: tools.EtcdErrorNotFound,
+	}
+
+	ctx := api.NewDefaultContext()
+	key := "foo"
+	path, err := etcdgeneric.NamespaceKeyFunc(ctx, "/registry/events", key)
+	if err != nil {
+		t.Errorf("Unexpected error: %v", err)
+	}
+
+	table := map[string]struct {
+		existing tools.EtcdResponseWithError
+		expect   tools.EtcdResponseWithError
+		toUpdate runtime.Object
+		errOK    func(error) bool
+	}{
+		"doesNotExist": {
+			existing: emptyNode,
+			expect:   nodeWithEventA,
+			toUpdate: eventA,
+			errOK:    func(err error) bool { return err == nil },
+		},
+		"doesNotExist2": {
+			existing: emptyNode,
+			expect:   nodeWithEventB,
+			toUpdate: eventB,
+			errOK:    func(err error) bool { return err == nil },
+		},
+		"replaceExisting": {
+			existing: nodeWithEventA,
+			expect:   nodeWithEventC,
+			toUpdate: eventC,
+			errOK:    func(err error) bool { return err == nil },
+		},
+	}
+
+	for name, item := range table {
+		fakeClient, registry := NewTestEventEtcdRegistry(t)
+		fakeClient.Data[path] = item.existing
+		err := registry.Update(ctx, 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, util.ObjectGoPrintDiff(e, a))
+		}
+	}
+}

pkg/registry/event/rest.go

@@ -66,6 +66,31 @@ func (rs *REST) Create(ctx api.Context, obj runtime.Object) (<-chan apiserver.RE
 	}), nil
 }
 
+// Update replaces an existing Event instance in storage.registry, with the given instance.
+func (rs *REST) Update(ctx api.Context, obj runtime.Object) (<-chan apiserver.RESTResult, error) {
+	event, ok := obj.(*api.Event)
+	if !ok {
+		return nil, fmt.Errorf("not an event object: %#v", obj)
+	}
+	if api.Namespace(ctx) != "" {
+		if !api.ValidNamespace(ctx, &event.ObjectMeta) {
+			return nil, errors.NewConflict("event", event.Namespace, fmt.Errorf("event.namespace does not match the provided context"))
+		}
+	}
+	if errs := validation.ValidateEvent(event); len(errs) > 0 {
+		return nil, errors.NewInvalid("event", event.Name, errs)
+	}
+	api.FillObjectMetaSystemFields(ctx, &event.ObjectMeta)
+
+	return apiserver.MakeAsync(func() (runtime.Object, error) {
+		err := rs.registry.Update(ctx, event.Name, event)
+		if err != nil {
+			return nil, err
+		}
+		return rs.registry.Get(ctx, event.Name)
+	}), nil
+}
+
 func (rs *REST) Delete(ctx api.Context, id string) (<-chan apiserver.RESTResult, error) {
 	obj, err := rs.registry.Get(ctx, id)
 	if err != nil {

pkg/registry/event/rest_test.go

@@ -97,6 +97,37 @@ func TestRESTCreate(t *testing.T) {
 	}
 }
 
+func TestRESTUpdate(t *testing.T) {
+	_, rest := NewTestREST()
+	eventA := testEvent("foo")
+	c, err := rest.Create(api.NewDefaultContext(), eventA)
+	if err != nil {
+		t.Fatalf("Unexpected error %v", err)
+	}
+	<-c
+	got, err := rest.Get(api.NewDefaultContext(), eventA.Name)
+	if err != nil {
+		t.Fatalf("Unexpected error %v", err)
+	}
+	if e, a := eventA, got; !reflect.DeepEqual(e, a) {
+		t.Errorf("diff: %s", util.ObjectDiff(e, a))
+	}
+	eventB := testEvent("bar")
+	u, err := rest.Update(api.NewDefaultContext(), eventB)
+	if err != nil {
+		t.Fatalf("Unexpected error %v", err)
+	}
+	<-u
+	got2, err := rest.Get(api.NewDefaultContext(), eventB.Name)
+	if err != nil {
+		t.Fatalf("Unexpected error %v", err)
+	}
+	if e, a := eventB, got2; !reflect.DeepEqual(e, a) {
+		t.Errorf("diff: %s", util.ObjectDiff(e, a))
+	}
+
+}
+
 func TestRESTDelete(t *testing.T) {
 	_, rest := NewTestREST()
 	eventA := testEvent("foo")

pkg/registry/generic/etcd/etcd.go

@@ -103,7 +103,7 @@ func (e *Etcd) Update(ctx api.Context, id string, obj runtime.Object) error {
 		return err
 	}
 	// TODO: verify that SetObj checks ResourceVersion before succeeding.
-	err = e.Helper.SetObj(key, obj)
+	err = e.Helper.SetObj(key, obj, 0 /* ttl */)
 	return etcderr.InterpretUpdateError(err, e.EndpointName, id)
 }
 

pkg/tools/etcd_tools.go

@@ -281,22 +281,22 @@ func (h *EtcdHelper) Delete(key string, recursive bool) error {
 	return err
 }
 
-// SetObj marshals obj via json, and stores under key. Will do an
+// SetObj marshals obj via json, and stores under key. Will do an atomic update if obj's ResourceVersion
-// atomic update if obj's ResourceVersion field is set.
+// field is set. 'ttl' is time-to-live in seconds, and 0 means forever.
-func (h *EtcdHelper) SetObj(key string, obj runtime.Object) error {
+func (h *EtcdHelper) SetObj(key string, obj runtime.Object, ttl uint64) error {
 	data, err := h.Codec.Encode(obj)
 	if err != nil {
 		return err
 	}
 	if h.ResourceVersioner != nil {
 		if version, err := h.ResourceVersioner.ResourceVersion(obj); err == nil && version != 0 {
-			_, err = h.Client.CompareAndSwap(key, string(data), 0, "", version)
+			_, err = h.Client.CompareAndSwap(key, string(data), ttl, "", version)
 			return err // err is shadowed!
 		}
 	}
 
 	// Create will fail if a key already exists.
-	_, err = h.Client.Create(key, string(data), 0)
+	_, err = h.Client.Create(key, string(data), ttl)
 	return err
 }
 

pkg/tools/etcd_tools_test.go

@@ -375,7 +375,7 @@ func TestSetObj(t *testing.T) {
 	obj := &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}}
 	fakeClient := NewFakeEtcdClient(t)
 	helper := EtcdHelper{fakeClient, testapi.Codec(), versioner}
-	err := helper.SetObj("/some/key", obj)
+	err := helper.SetObj("/some/key", obj, 5)
 	if err != nil {
 		t.Errorf("Unexpected error %#v", err)
 	}
@@ -388,6 +388,10 @@ func TestSetObj(t *testing.T) {
 	if expect != got {
 		t.Errorf("Wanted %v, got %v", expect, got)
 	}
+	if e, a := uint64(5), fakeClient.LastSetTTL; e != a {
+		t.Errorf("Wanted %v, got %v", e, a)
+	}
+
 }
 
 func TestSetObjWithVersion(t *testing.T) {
@@ -404,7 +408,7 @@ func TestSetObjWithVersion(t *testing.T) {
 	}
 
 	helper := EtcdHelper{fakeClient, testapi.Codec(), versioner}
-	err := helper.SetObj("/some/key", obj)
+	err := helper.SetObj("/some/key", obj, 7)
 	if err != nil {
 		t.Fatalf("Unexpected error %#v", err)
 	}
@@ -417,13 +421,16 @@ func TestSetObjWithVersion(t *testing.T) {
 	if expect != got {
 		t.Errorf("Wanted %v, got %v", expect, got)
 	}
+	if e, a := uint64(7), fakeClient.LastSetTTL; e != a {
+		t.Errorf("Wanted %v, got %v", e, a)
+	}
 }
 
 func TestSetObjWithoutResourceVersioner(t *testing.T) {
 	obj := &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}}
 	fakeClient := NewFakeEtcdClient(t)
 	helper := EtcdHelper{fakeClient, testapi.Codec(), nil}
-	err := helper.SetObj("/some/key", obj)
+	err := helper.SetObj("/some/key", obj, 3)
 	if err != nil {
 		t.Errorf("Unexpected error %#v", err)
 	}
@@ -436,6 +443,9 @@ func TestSetObjWithoutResourceVersioner(t *testing.T) {
 	if expect != got {
 		t.Errorf("Wanted %v, got %v", expect, got)
 	}
+	if e, a := uint64(3), fakeClient.LastSetTTL; e != a {
+		t.Errorf("Wanted %v, got %v", e, a)
+	}
 }
 
 func TestAtomicUpdate(t *testing.T) {

pkg/tools/fake_etcd_client.go

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

test/integration/auth_test.go

@@ -24,12 +24,14 @@ package integration
 
 import (
 	"bytes"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"io/ioutil"
 	"net/http"
 	"net/http/httptest"
 	"os"
+	"strings"
 	"testing"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/apiserver"
@@ -73,7 +75,7 @@ var aPod string = `
       "id": "a",
       "containers": [{ "name": "foo", "image": "bar/foo", }]
     }
-  },
+  }%s
 }
 `
 var aRC string = `
@@ -97,7 +99,7 @@ var aRC string = `
        },
        "labels": {"name": "a"}
       }},
-  "labels": {"name": "a"}
+  "labels": {"name": "a"}%s
 }
 `
 var aService string = `
@@ -108,7 +110,7 @@ var aService string = `
   "port": 8000,
   "portalIP": "10.0.0.100",
   "labels": { "name": "a" },
-  "selector": { "name": "a" }
+  "selector": { "name": "a" }%s
 }
 `
 var aMinion string = `
@@ -116,7 +118,7 @@ var aMinion string = `
   "kind": "Minion",
   "apiVersion": "v1beta1",
   "id": "a",
-  "hostIP": "10.10.10.10",
+  "hostIP": "10.10.10.10"%s
 }
 `
 
@@ -131,7 +133,7 @@ var aEvent string = `
     "name": "a",
     "namespace": "default",
     "apiVersion": "v1beta1",
-  }
+  }%s
 }
 `
 
@@ -141,7 +143,7 @@ var aBinding string = `
   "apiVersion": "v1beta1",
   "id": "a",
   "host": "10.10.10.10",
-  "podID": "a"
+  "podID": "a"%s
 }
 `
 
@@ -150,7 +152,7 @@ var aEndpoints string = `
   "kind": "Endpoints",
   "apiVersion": "v1beta1",
   "id": "a",
-  "endpoints": ["10.10.1.1:1909"],
+  "endpoints": ["10.10.1.1:1909"]%s
 }
 `
 
@@ -183,7 +185,7 @@ func getTestRequests() []struct {
 		// Normal methods on pods
 		{"GET", "/api/v1beta1/pods", "", code200},
 		{"POST", "/api/v1beta1/pods" + timeoutFlag, aPod, code200},
-		{"PUT", "/api/v1beta1/pods/a" + timeoutFlag, aPod, code500}, // See #2114 about why 500
+		{"PUT", "/api/v1beta1/pods/a" + timeoutFlag, aPod, code200},
 		{"GET", "/api/v1beta1/pods", "", code200},
 		{"GET", "/api/v1beta1/pods/a", "", code200},
 		{"DELETE", "/api/v1beta1/pods/a" + timeoutFlag, "", code200},
@@ -203,7 +205,7 @@ func getTestRequests() []struct {
 		// Normal methods on services
 		{"GET", "/api/v1beta1/services", "", code200},
 		{"POST", "/api/v1beta1/services" + timeoutFlag, aService, code200},
-		{"PUT", "/api/v1beta1/services/a" + timeoutFlag, aService, code409}, // See #2115 about why 409
+		{"PUT", "/api/v1beta1/services/a" + timeoutFlag, aService, code200},
 		{"GET", "/api/v1beta1/services", "", code200},
 		{"GET", "/api/v1beta1/services/a", "", code200},
 		{"DELETE", "/api/v1beta1/services/a" + timeoutFlag, "", code200},
@@ -211,7 +213,7 @@ func getTestRequests() []struct {
 		// Normal methods on replicationControllers
 		{"GET", "/api/v1beta1/replicationControllers", "", code200},
 		{"POST", "/api/v1beta1/replicationControllers" + timeoutFlag, aRC, code200},
-		{"PUT", "/api/v1beta1/replicationControllers/a" + timeoutFlag, aRC, code409}, // See #2115 about why 409
+		{"PUT", "/api/v1beta1/replicationControllers/a" + timeoutFlag, aRC, code200},
 		{"GET", "/api/v1beta1/replicationControllers", "", code200},
 		{"GET", "/api/v1beta1/replicationControllers/a", "", code200},
 		{"DELETE", "/api/v1beta1/replicationControllers/a" + timeoutFlag, "", code200},
@@ -235,7 +237,7 @@ func getTestRequests() []struct {
 		// Normal methods on events
 		{"GET", "/api/v1beta1/events", "", code200},
 		{"POST", "/api/v1beta1/events" + timeoutFlag, aEvent, code200},
-		{"PUT", "/api/v1beta1/events/a" + timeoutFlag, aEvent, code405},
+		{"PUT", "/api/v1beta1/events/a" + timeoutFlag, aEvent, code200},
 		{"GET", "/api/v1beta1/events", "", code200},
 		{"GET", "/api/v1beta1/events", "", code200},
 		{"GET", "/api/v1beta1/events/a", "", code200},
@@ -308,10 +310,22 @@ func TestAuthModeAlwaysAllow(t *testing.T) {
 	})
 
 	transport := http.DefaultTransport
+	previousResourceVersion := make(map[string]float64)
 
 	for _, r := range getTestRequests() {
 		t.Logf("case %v", r)
-		bodyBytes := bytes.NewReader([]byte(r.body))
+		var bodyStr string
+		if r.body != "" {
+			bodyStr = fmt.Sprintf(r.body, "")
+			if r.verb == "PUT" && r.body != "" {
+				// For update operations, insert previous resource version
+				if resVersion := previousResourceVersion[getPreviousResourceVersionKey(r.URL, "")]; resVersion != 0 {
+					resourceVersionJson := fmt.Sprintf(",\r\n\"resourceVersion\": %v", resVersion)
+					bodyStr = fmt.Sprintf(r.body, resourceVersionJson)
+				}
+			}
+		}
+		bodyBytes := bytes.NewReader([]byte(bodyStr))
 		req, err := http.NewRequest(r.verb, s.URL+r.URL, bodyBytes)
 		if err != nil {
 			t.Fatalf("unexpected error: %v", err)
@@ -322,15 +336,50 @@ func TestAuthModeAlwaysAllow(t *testing.T) {
 			if err != nil {
 				t.Fatalf("unexpected error: %v", err)
 			}
+			b, _ := ioutil.ReadAll(resp.Body)
 			if _, ok := r.statusCodes[resp.StatusCode]; !ok {
 				t.Errorf("Expected status one of %v, but got %v", r.statusCodes, resp.StatusCode)
-				b, _ := ioutil.ReadAll(resp.Body)
 				t.Errorf("Body: %v", string(b))
+			} else {
+				if r.verb == "POST" {
+					// For successful create operations, extract resourceVersion
+					id, currentResourceVersion, err := parseResourceVersion(b)
+					if err == nil {
+						key := getPreviousResourceVersionKey(r.URL, id)
+						previousResourceVersion[key] = currentResourceVersion
+					}
+				}
 			}
 		}()
 	}
 }
 
+func parseResourceVersion(response []byte) (string, float64, error) {
+	var resultBodyMap map[string]interface{}
+	err := json.Unmarshal(response, &resultBodyMap)
+	if err != nil {
+		return "", 0, fmt.Errorf("unexpected error unmarshaling resultBody: %v", err)
+	}
+	id, ok := resultBodyMap["id"].(string)
+	if !ok {
+		return "", 0, fmt.Errorf("unexpected error, id not found in JSON response: %v", string(response))
+	}
+	resourceVersion, ok := resultBodyMap["resourceVersion"].(float64)
+	if !ok {
+		return "", 0, fmt.Errorf("unexpected error, resourceVersion not found in JSON response: %v", string(response))
+	}
+	return id, resourceVersion, nil
+}
+
+func getPreviousResourceVersionKey(url, id string) string {
+	baseUrl := strings.Split(url, "?")[0]
+	key := baseUrl
+	if id != "" {
+		key = fmt.Sprintf("%s/%v", baseUrl, id)
+	}
+	return key
+}
+
 func TestAuthModeAlwaysDeny(t *testing.T) {
 	deleteAllEtcdKeys()
 
@@ -426,12 +475,24 @@ func TestAliceNotForbiddenOrUnauthorized(t *testing.T) {
 		AdmissionControl:  admit.NewAlwaysAdmit(),
 	})
 
+	previousResourceVersion := make(map[string]float64)
 	transport := http.DefaultTransport
 
 	for _, r := range getTestRequests() {
 		token := AliceToken
 		t.Logf("case %v", r)
-		bodyBytes := bytes.NewReader([]byte(r.body))
+		var bodyStr string
+		if r.body != "" {
+			bodyStr = fmt.Sprintf(r.body, "")
+			if r.verb == "PUT" && r.body != "" {
+				// For update operations, insert previous resource version
+				if resVersion := previousResourceVersion[getPreviousResourceVersionKey(r.URL, "")]; resVersion != 0 {
+					resourceVersionJson := fmt.Sprintf(",\r\n\"resourceVersion\": %v", resVersion)
+					bodyStr = fmt.Sprintf(r.body, resourceVersionJson)
+				}
+			}
+		}
+		bodyBytes := bytes.NewReader([]byte(bodyStr))
 		req, err := http.NewRequest(r.verb, s.URL+r.URL, bodyBytes)
 		if err != nil {
 			t.Fatalf("unexpected error: %v", err)
@@ -444,11 +505,21 @@ func TestAliceNotForbiddenOrUnauthorized(t *testing.T) {
 			if err != nil {
 				t.Fatalf("unexpected error: %v", err)
 			}
+			b, _ := ioutil.ReadAll(resp.Body)
 			if _, ok := r.statusCodes[resp.StatusCode]; !ok {
 				t.Errorf("Expected status one of %v, but got %v", r.statusCodes, resp.StatusCode)
-				b, _ := ioutil.ReadAll(resp.Body)
 				t.Errorf("Body: %v", string(b))
+			} else {
+				if r.verb == "POST" {
+					// For successful create operations, extract resourceVersion
+					id, currentResourceVersion, err := parseResourceVersion(b)
+					if err == nil {
+						key := getPreviousResourceVersionKey(r.URL, id)
+						previousResourceVersion[key] = currentResourceVersion
+					}
+				}
 			}
+
 		}()
 	}
 }
@@ -628,6 +699,7 @@ func TestNamespaceAuthorization(t *testing.T) {
 		AdmissionControl:  admit.NewAlwaysAdmit(),
 	})
 
+	previousResourceVersion := make(map[string]float64)
 	transport := http.DefaultTransport
 
 	requests := []struct {
@@ -656,7 +728,18 @@ func TestNamespaceAuthorization(t *testing.T) {
 	for _, r := range requests {
 		token := BobToken
 		t.Logf("case %v", r)
-		bodyBytes := bytes.NewReader([]byte(r.body))
+		var bodyStr string
+		if r.body != "" {
+			bodyStr = fmt.Sprintf(r.body, "")
+			if r.verb == "PUT" && r.body != "" {
+				// For update operations, insert previous resource version
+				if resVersion := previousResourceVersion[getPreviousResourceVersionKey(r.URL, "")]; resVersion != 0 {
+					resourceVersionJson := fmt.Sprintf(",\r\n\"resourceVersion\": %v", resVersion)
+					bodyStr = fmt.Sprintf(r.body, resourceVersionJson)
+				}
+			}
+		}
+		bodyBytes := bytes.NewReader([]byte(bodyStr))
 		req, err := http.NewRequest(r.verb, s.URL+r.URL, bodyBytes)
 		if err != nil {
 			t.Fatalf("unexpected error: %v", err)
@@ -668,11 +751,21 @@ func TestNamespaceAuthorization(t *testing.T) {
 			if err != nil {
 				t.Fatalf("unexpected error: %v", err)
 			}
+			b, _ := ioutil.ReadAll(resp.Body)
 			if _, ok := r.statusCodes[resp.StatusCode]; !ok {
 				t.Errorf("Expected status one of %v, but got %v", r.statusCodes, resp.StatusCode)
-				b, _ := ioutil.ReadAll(resp.Body)
 				t.Errorf("Body: %v", string(b))
+			} else {
+				if r.verb == "POST" {
+					// For successful create operations, extract resourceVersion
+					id, currentResourceVersion, err := parseResourceVersion(b)
+					if err == nil {
+						key := getPreviousResourceVersionKey(r.URL, id)
+						previousResourceVersion[key] = currentResourceVersion
+					}
+				}
 			}
+
 		}()
 	}
 }
@@ -713,6 +806,7 @@ func TestKindAuthorization(t *testing.T) {
 		AdmissionControl:  admit.NewAlwaysAdmit(),
 	})
 
+	previousResourceVersion := make(map[string]float64)
 	transport := http.DefaultTransport
 
 	requests := []struct {
@@ -735,7 +829,18 @@ func TestKindAuthorization(t *testing.T) {
 	for _, r := range requests {
 		token := BobToken
 		t.Logf("case %v", r)
-		bodyBytes := bytes.NewReader([]byte(r.body))
+		var bodyStr string
+		if r.body != "" {
+			bodyStr = fmt.Sprintf(r.body, "")
+			if r.verb == "PUT" && r.body != "" {
+				// For update operations, insert previous resource version
+				if resVersion := previousResourceVersion[getPreviousResourceVersionKey(r.URL, "")]; resVersion != 0 {
+					resourceVersionJson := fmt.Sprintf(",\r\n\"resourceVersion\": %v", resVersion)
+					bodyStr = fmt.Sprintf(r.body, resourceVersionJson)
+				}
+			}
+		}
+		bodyBytes := bytes.NewReader([]byte(bodyStr))
 		req, err := http.NewRequest(r.verb, s.URL+r.URL, bodyBytes)
 		if err != nil {
 			t.Fatalf("unexpected error: %v", err)
@@ -747,11 +852,21 @@ func TestKindAuthorization(t *testing.T) {
 			if err != nil {
 				t.Fatalf("unexpected error: %v", err)
 			}
+			b, _ := ioutil.ReadAll(resp.Body)
 			if _, ok := r.statusCodes[resp.StatusCode]; !ok {
 				t.Errorf("Expected status one of %v, but got %v", r.statusCodes, resp.StatusCode)
-				b, _ := ioutil.ReadAll(resp.Body)
 				t.Errorf("Body: %v", string(b))
+			} else {
+				if r.verb == "POST" {
+					// For successful create operations, extract resourceVersion
+					id, currentResourceVersion, err := parseResourceVersion(b)
+					if err == nil {
+						key := getPreviousResourceVersionKey(r.URL, id)
+						previousResourceVersion[key] = currentResourceVersion
+					}
+				}
 			}
+
 		}
 	}
 }

test/integration/etcd_tools_test.go

@@ -60,7 +60,7 @@ func TestSetObj(t *testing.T) {
 	helper := tools.EtcdHelper{Client: client, Codec: stringCodec{}}
 	withEtcdKey(func(key string) {
 		fakeObject := fakeAPIObject("object")
-		if err := helper.SetObj(key, &fakeObject); err != nil {
+		if err := helper.SetObj(key, &fakeObject, 0 /* ttl */); err != nil {
 			t.Fatalf("unexpected error: %v", err)
 		}
 		resp, err := client.Get(key, false, false)