Send correct resource version for delete events from watch cache

staging/src/k8s.io/apiserver/pkg/storage/BUILD

@@ -21,6 +21,7 @@ go_test(
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/api/equality:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/api/meta:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/fields:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/labels:go_default_library",

staging/src/k8s.io/apiserver/pkg/storage/cacher.go

@@ -338,7 +338,7 @@ func (c *Cacher) Watch(ctx context.Context, key string, resourceVersion string,
 	c.Lock()
 	defer c.Unlock()
 	forget := forgetWatcher(c, c.watcherIdx, triggerValue, triggerSupported)
-	watcher := newCacheWatcher(watchRV, chanSize, initEvents, watchFilterFunction(key, pred), forget)
+	watcher := newCacheWatcher(watchRV, chanSize, initEvents, watchFilterFunction(key, pred), forget, c.versioner)
 
 	c.watchers.addWatcher(watcher, c.watcherIdx, triggerValue, triggerSupported)
 	c.watcherIdx++
@@ -796,9 +796,10 @@ type cacheWatcher struct {
 	filter    watchFilterFunc
 	stopped   bool
 	forget    func(bool)
+	versioner Versioner
 }
 
-func newCacheWatcher(resourceVersion uint64, chanSize int, initEvents []*watchCacheEvent, filter watchFilterFunc, forget func(bool)) *cacheWatcher {
+func newCacheWatcher(resourceVersion uint64, chanSize int, initEvents []*watchCacheEvent, filter watchFilterFunc, forget func(bool), versioner Versioner) *cacheWatcher {
 	watcher := &cacheWatcher{
 		input:     make(chan *watchCacheEvent, chanSize),
 		result:    make(chan watch.Event, chanSize),
@@ -806,6 +807,7 @@ func newCacheWatcher(resourceVersion uint64, chanSize int, initEvents []*watchCa
 		filter:    filter,
 		stopped:   false,
 		forget:    forget,
+		versioner: versioner,
 	}
 	go watcher.process(initEvents, resourceVersion)
 	return watcher
@@ -894,7 +896,12 @@ func (c *cacheWatcher) sendWatchCacheEvent(event *watchCacheEvent) {
 	case curObjPasses && oldObjPasses:
 		watchEvent = watch.Event{Type: watch.Modified, Object: event.Object.DeepCopyObject()}
 	case !curObjPasses && oldObjPasses:
-		watchEvent = watch.Event{Type: watch.Deleted, Object: event.PrevObject.DeepCopyObject()}
+		// return a delete event with the previous object content, but with the event's resource version
+		oldObj := event.PrevObject.DeepCopyObject()
+		if err := c.versioner.UpdateObject(oldObj, event.ResourceVersion); err != nil {
+			utilruntime.HandleError(fmt.Errorf("failure to version api object (%d) %#v: %v", event.ResourceVersion, oldObj, err))
+		}
+		watchEvent = watch.Event{Type: watch.Deleted, Object: oldObj}
 	}
 
 	// We need to ensure that if we put event X to the c.result, all

staging/src/k8s.io/apiserver/pkg/storage/cacher_whitebox_test.go

@@ -17,15 +17,19 @@ limitations under the License.
 package storage
 
 import (
+	"fmt"
 	"reflect"
+	"strconv"
 	"sync"
 	"testing"
 	"time"
 
 	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/diff"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apimachinery/pkg/watch"
@@ -48,7 +52,7 @@ func TestCacheWatcherCleanupNotBlockedByResult(t *testing.T) {
 	}
 	// set the size of the buffer of w.result to 0, so that the writes to
 	// w.result is blocked.
-	w := newCacheWatcher(0, 0, initEvents, filter, forget)
+	w := newCacheWatcher(0, 0, initEvents, filter, forget, testVersioner{})
 	w.Stop()
 	if err := wait.PollImmediate(1*time.Second, 5*time.Second, func() (bool, error) {
 		lock.RLock()
@@ -76,6 +80,7 @@ func TestCacheWatcherHandlesFiltering(t *testing.T) {
 					Type:            watch.Added,
 					Object:          &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "1"}},
 					ObjFields:       fields.Set{"spec.nodeName": "host"},
+					ResourceVersion: 1,
 				},
 				{
 					Type:            watch.Modified,
@@ -83,6 +88,7 @@ func TestCacheWatcherHandlesFiltering(t *testing.T) {
 					PrevObjFields:   fields.Set{"spec.nodeName": "host"},
 					Object:          &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "2"}},
 					ObjFields:       fields.Set{"spec.nodeName": ""},
+					ResourceVersion: 2,
 				},
 				{
 					Type:            watch.Modified,
@@ -90,11 +96,12 @@ func TestCacheWatcherHandlesFiltering(t *testing.T) {
 					PrevObjFields:   fields.Set{"spec.nodeName": ""},
 					Object:          &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "3"}},
 					ObjFields:       fields.Set{"spec.nodeName": "host"},
+					ResourceVersion: 3,
 				},
 			},
 			expected: []watch.Event{
 				{Type: watch.Added, Object: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "1"}}},
-				{Type: watch.Deleted, Object: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "1"}}},
+				{Type: watch.Deleted, Object: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "2"}}},
 				{Type: watch.Added, Object: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "3"}}},
 			},
 		},
@@ -105,6 +112,7 @@ func TestCacheWatcherHandlesFiltering(t *testing.T) {
 					Type:            watch.Added,
 					Object:          &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "1"}},
 					ObjFields:       fields.Set{"spec.nodeName": ""},
+					ResourceVersion: 1,
 				},
 				{
 					Type:            watch.Modified,
@@ -112,6 +120,7 @@ func TestCacheWatcherHandlesFiltering(t *testing.T) {
 					PrevObjFields:   fields.Set{"spec.nodeName": ""},
 					Object:          &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "2"}},
 					ObjFields:       fields.Set{"spec.nodeName": ""},
+					ResourceVersion: 2,
 				},
 				{
 					Type:            watch.Modified,
@@ -119,6 +128,7 @@ func TestCacheWatcherHandlesFiltering(t *testing.T) {
 					PrevObjFields:   fields.Set{"spec.nodeName": ""},
 					Object:          &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "3"}},
 					ObjFields:       fields.Set{"spec.nodeName": "host"},
+					ResourceVersion: 3,
 				},
 				{
 					Type:            watch.Modified,
@@ -126,6 +136,7 @@ func TestCacheWatcherHandlesFiltering(t *testing.T) {
 					PrevObjFields:   fields.Set{"spec.nodeName": "host"},
 					Object:          &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "4"}},
 					ObjFields:       fields.Set{"spec.nodeName": "host"},
+					ResourceVersion: 4,
 				},
 				{
 					Type:            watch.Modified,
@@ -133,6 +144,7 @@ func TestCacheWatcherHandlesFiltering(t *testing.T) {
 					PrevObjFields:   fields.Set{"spec.nodeName": "host"},
 					Object:          &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "5"}},
 					ObjFields:       fields.Set{"spec.nodeName": ""},
+					ResourceVersion: 5,
 				},
 				{
 					Type:            watch.Modified,
@@ -140,12 +152,13 @@ func TestCacheWatcherHandlesFiltering(t *testing.T) {
 					PrevObjFields:   fields.Set{"spec.nodeName": ""},
 					Object:          &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "6"}},
 					ObjFields:       fields.Set{"spec.nodeName": ""},
+					ResourceVersion: 6,
 				},
 			},
 			expected: []watch.Event{
 				{Type: watch.Added, Object: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "3"}}},
 				{Type: watch.Modified, Object: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "4"}}},
-				{Type: watch.Deleted, Object: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "4"}}},
+				{Type: watch.Deleted, Object: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "5"}}},
 			},
 		},
 	}
@@ -157,7 +170,7 @@ TestCase:
 		for j := range testCase.events {
 			testCase.events[j].ResourceVersion = uint64(j) + 1
 		}
-		w := newCacheWatcher(0, 0, testCase.events, filter, forget)
+		w := newCacheWatcher(0, 0, testCase.events, filter, forget, testVersioner{})
 		ch := w.ResultChan()
 		for j, event := range testCase.expected {
 			e := <-ch
@@ -175,3 +188,18 @@ TestCase:
 		w.Stop()
 	}
 }
+
+type testVersioner struct{}
+
+func (testVersioner) UpdateObject(obj runtime.Object, resourceVersion uint64) error {
+	return meta.NewAccessor().SetResourceVersion(obj, strconv.FormatUint(resourceVersion, 10))
+}
+func (testVersioner) UpdateList(obj runtime.Object, resourceVersion uint64, continueValue string) error {
+	return fmt.Errorf("unimplemented")
+}
+func (testVersioner) PrepareObjectForStorage(obj runtime.Object) error {
+	return fmt.Errorf("unimplemented")
+}
+func (testVersioner) ObjectResourceVersion(obj runtime.Object) (uint64, error) {
+	return 0, fmt.Errorf("unimplemented")
+}