handle watch for unsafe delete

Author: tkashem

staging/src/k8s.io/apiserver/pkg/storage/etcd3/errors.go

@@ -17,7 +17,11 @@ limitations under the License.
 package etcd3
 
 import (
+	goerrors "errors"
+	"net/http"
+
 	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apiserver/pkg/storage"
 
 	etcdrpc "go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
@@ -29,6 +33,19 @@ func interpretWatchError(err error) error {
 	case err == etcdrpc.ErrCompacted:
 		return errors.NewResourceExpired("The resourceVersion for the provided watch is too old.")
 	}
+
+	var corruptobjDeletedErr *corruptObjectDeletedError
+	if goerrors.As(err, &corruptobjDeletedErr) {
+		return &errors.StatusError{
+			ErrStatus: metav1.Status{
+				Status:  metav1.StatusFailure,
+				Code:    http.StatusInternalServerError,
+				Reason:  metav1.StatusReasonStoreReadError,
+				Message: corruptobjDeletedErr.Error(),
+			},
+		}
+	}
+
 	return err
 }
 

staging/src/k8s.io/apiserver/pkg/storage/etcd3/store_test.go

@@ -194,6 +194,28 @@ func (s *storeWithPrefixTransformer) UpdatePrefixTransformer(modifier storagetes
 	}
 }
 
+type corruptedTransformer struct {
+	value.Transformer
+}
+
+func (f *corruptedTransformer) TransformFromStorage(ctx context.Context, data []byte, dataCtx value.Context) (out []byte, stale bool, err error) {
+	return nil, true, &corruptObjectError{err: fmt.Errorf("bits flipped"), errType: untransformable}
+}
+
+type storeWithCorruptedTransformer struct {
+	*store
+}
+
+func (s *storeWithCorruptedTransformer) CorruptTransformer() func() {
+	ct := &corruptedTransformer{Transformer: s.transformer}
+	s.transformer = ct
+	s.watcher.transformer = ct
+	return func() {
+		s.transformer = ct.Transformer
+		s.watcher.transformer = ct.Transformer
+	}
+}
+
 func TestGuaranteedUpdate(t *testing.T) {
 	ctx, store, client := testSetup(t)
 	storagetesting.RunTestGuaranteedUpdate(ctx, t, &storeWithPrefixTransformer{store}, checkStorageInvariants(client.Client, store.codec))

staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher.go

@@ -686,18 +686,40 @@ func (wc *watchChan) prepareObjs(e *event) (curObj runtime.Object, oldObj runtim
 	if len(e.prevValue) > 0 && (e.isDeleted || !wc.acceptAll()) {
 		data, _, err := wc.watcher.transformer.TransformFromStorage(wc.ctx, e.prevValue, authenticatedDataString(e.key))
 		if err != nil {
-			return nil, nil, err
+			return nil, nil, wc.watcher.transformIfCorruptObjectError(e, err)
 		}
 		// Note that this sends the *old* object with the etcd revision for the time at
 		// which it gets deleted.
 		oldObj, err = decodeObj(wc.watcher.codec, wc.watcher.versioner, data, e.rev)
 		if err != nil {
-			return nil, nil, err
+			return nil, nil, wc.watcher.transformIfCorruptObjectError(e, err)
 		}
 	}
 	return curObj, oldObj, nil
 }
 
+type corruptObjectDeletedError struct {
+	err error
+}
+
+func (e *corruptObjectDeletedError) Error() string {
+	return fmt.Sprintf("saw a DELETED event, but object data is corrupt - %v", e.err)
+}
+func (e *corruptObjectDeletedError) Unwrap() error { return e.err }
+
+func (w *watcher) transformIfCorruptObjectError(e *event, err error) error {
+	var corruptObjErr *corruptObjectError
+	if !e.isDeleted || !errors.As(err, &corruptObjErr) {
+		return err
+	}
+
+	// if we are here it means we received a DELETED event but the object
+	// associated with it is corrupt because we failed to transform or
+	// decode the data associated with the object.
+	// wrap the original error so we can send a proper watch Error event.
+	return &corruptObjectDeletedError{err: corruptObjErr}
+}
+
 func decodeObj(codec runtime.Codec, versioner storage.Versioner, data []byte, rev int64) (_ runtime.Object, err error) {
 	obj, err := runtime.Decode(codec, []byte(data))
 	if err != nil {

staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher_test.go

@@ -112,6 +112,12 @@ func TestProgressNotify(t *testing.T) {
 	storagetesting.RunOptionalTestProgressNotify(ctx, t, store)
 }
 
+func TestWatchWithUnsafeDelete(t *testing.T) {
+	featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.AllowUnsafeMalformedObjectDeletion, true)
+	ctx, store, _ := testSetup(t)
+	storagetesting.RunTestWatchWithUnsafeDelete(ctx, t, &storeWithCorruptedTransformer{store})
+}
+
 // TestWatchDispatchBookmarkEvents makes sure that
 // setting allowWatchBookmarks query param against
 // etcd implementation doesn't have any effect.

staging/src/k8s.io/apiserver/pkg/storage/testing/store_tests.go

@@ -2147,6 +2147,11 @@ type InterfaceWithPrefixTransformer interface {
 	UpdatePrefixTransformer(PrefixTransformerModifier) func()
 }
 
+type InterfaceWithCorruptTransformer interface {
+	storage.Interface
+	CorruptTransformer() func()
+}
+
 func RunTestListResourceVersionMatch(ctx context.Context, t *testing.T, store InterfaceWithPrefixTransformer) {
 	nextPod := func(index uint32) (string, *example.Pod) {
 		obj := &example.Pod{

staging/src/k8s.io/apiserver/pkg/storage/testing/watcher_tests.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -407,6 +408,61 @@ func RunTestWatchError(ctx context.Context, t *testing.T, store InterfaceWithPre
 	testCheckEventType(t, w, watch.Error)
 }
 
+func RunTestWatchWithUnsafeDelete(ctx context.Context, t *testing.T, store InterfaceWithCorruptTransformer) {
+	obj := &example.Pod{ObjectMeta: metav1.ObjectMeta{Name: "foo", Namespace: "test-ns"}}
+	key := computePodKey(obj)
+
+	out := &example.Pod{}
+	if err := store.Create(ctx, key, obj, out, 0); err != nil {
+		t.Fatalf("failed to create object in the store: %v", err)
+	}
+
+	// Compute the initial resource version from which we can start watching later.
+	list := &example.PodList{}
+	storageOpts := storage.ListOptions{
+		ResourceVersion: "0",
+		Predicate:       storage.Everything,
+		Recursive:       true,
+	}
+	if err := store.GetList(ctx, "/pods", storageOpts, list); err != nil {
+		t.Errorf("Unexpected error: %v", err)
+	}
+
+	// Now trigger watch error by injecting failing transformer.
+	revertTransformer := store.CorruptTransformer()
+	defer revertTransformer()
+
+	w, err := store.Watch(ctx, key, storage.ListOptions{ResourceVersion: list.ResourceVersion, Predicate: storage.Everything})
+	if err != nil {
+		t.Fatalf("Watch failed: %v", err)
+	}
+
+	// normal deletetion should fail
+	if err := store.Delete(ctx, key, &example.Pod{}, nil, storage.ValidateAllObjectFunc, nil, storage.DeleteOptions{}); err == nil {
+		t.Fatalf("Expected normal Delete to fail")
+	}
+	if err := store.Delete(ctx, key, &example.Pod{}, nil, storage.ValidateAllObjectFunc, nil, storage.DeleteOptions{IgnoreStoreReadError: true}); err != nil {
+		t.Fatalf("Expected unsafe Delete to succeed, but got: %v", err)
+	}
+
+	testCheckResultFunc(t, w, func(got watch.Event) {
+		if want, got := watch.Error, got.Type; want != got {
+			t.Errorf("Expected event type: %q, but got: %q", want, got)
+		}
+		switch v := got.Object.(type) {
+		case *metav1.Status:
+			if want, got := metav1.StatusReasonStoreReadError, v.Reason; want != got {
+				t.Errorf("Expected reason: %q, but got: %q", want, got)
+			}
+			if want := "saw a DELETED event, but object data is corrupt"; !strings.Contains(v.Message, want) {
+				t.Errorf("Expected Message to contain: %q, but got: %q", want, v.Message)
+			}
+		default:
+			t.Errorf("expected an metav1 Status object, but got: %v", got.Object)
+		}
+	})
+}
+
 func RunTestWatchContextCancel(ctx context.Context, t *testing.T, store storage.Interface) {
 	canceledCtx, cancel := context.WithCancel(ctx)
 	cancel()

staging/src/k8s.io/client-go/tools/cache/reflector_test.go

@@ -21,9 +21,12 @@ import (
 	"errors"
 	"fmt"
 	"math/rand"
+	"net/http"
 	"reflect"
 	goruntime "runtime"
 	"strconv"
+	"sync"
+	"sync/atomic"
 	"syscall"
 	"testing"
 	"time"
@@ -1753,6 +1756,152 @@ func TestReflectorListExtract(t *testing.T) {
 	}
 }
 
+func TestReflectorReplacesStoreOnUnsafeDelete(t *testing.T) {
+	mkPod := func(id string, rv string) *v1.Pod {
+		return &v1.Pod{ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: id, ResourceVersion: rv}}
+	}
+	mkList := func(rv string, pods ...*v1.Pod) *v1.PodList {
+		list := &v1.PodList{ListMeta: metav1.ListMeta{ResourceVersion: rv}}
+		for _, pod := range pods {
+			list.Items = append(list.Items, *pod)
+		}
+		return list
+	}
+	makeStatus := func() *metav1.Status {
+		return &metav1.Status{
+			Status:  metav1.StatusFailure,
+			Code:    http.StatusInternalServerError,
+			Reason:  metav1.StatusReasonStoreReadError,
+			Message: "failed to prepare current and previous objects: corrupt object has been deleted",
+		}
+	}
+
+	// these pods preexist and never get updated/deleted
+	preExisting := mkPod("foo-1", "1")
+	pods := []*v1.Pod{preExisting, mkPod("foo-2", "2"), mkPod("foo-3", "3")}
+	lastExpectedRV := "5"
+	lists := []*v1.PodList{
+		mkList("3", pods...),            // initial list
+		mkList(lastExpectedRV, pods...), // re-list due to watch error
+	}
+	corruptObj := mkPod("foo", "4")
+	events := []watch.Event{
+		{Type: watch.Added, Object: corruptObj},
+		// the object becomes corrupt, and it gets unsafe-deleted, and
+		// watch sends the following Error event, note the RV has
+		// advanced to "5" in the storage due to the delete operation
+		{Type: watch.Error, Object: makeStatus()},
+	}
+
+	s := NewFIFO(MetaNamespaceKeyFunc)
+	var replaceInvoked atomic.Int32
+	store := &fakeStore{
+		Store: s,
+		beforeReplace: func(list []interface{}, rv string) {
+			// interested in the Replace call that happens after the Error event
+			if rv == lastExpectedRV {
+				replaceInvoked.Add(1)
+				_, exists, err := s.Get(corruptObj)
+				if err != nil || !exists {
+					t.Errorf("expected the object to exist in the store, exists: %t, err: %v", exists, err)
+				}
+				_, exists, err = s.Get(preExisting)
+				if err != nil || !exists {
+					t.Errorf("expected the pre-existing object to be in the store, exists: %t, err: %v", exists, err)
+				}
+			}
+		},
+		afterReplace: func(rv string, err error) {
+			if rv == lastExpectedRV {
+				replaceInvoked.Add(1)
+				if err != nil {
+					t.Errorf("expected Replace to have succeeded, but got error: %v", err)
+				}
+				_, exists, err := s.Get(corruptObj)
+				if err != nil || exists {
+					t.Errorf("expected the object to have been removed from the store, exists: %t, err: %v", exists, err)
+				}
+				// show that a pre-existing pod is still in the cache
+				_, exists, err = s.Get(preExisting)
+				if err != nil || !exists {
+					t.Errorf("expected the pre-existing object to be in the store, exists: %t, err: %v", exists, err)
+				}
+			}
+		},
+	}
+
+	var once sync.Once
+	lw := &testLW{
+		WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
+			fw := watch.NewFake()
+			go func() {
+				once.Do(func() {
+					for _, e := range events {
+						fw.Action(e.Type, e.Object)
+					}
+				})
+			}()
+			return fw, nil
+		},
+		ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
+			var list runtime.Object
+			if len(lists) > 0 {
+				list = lists[0]
+				lists = lists[1:]
+			}
+			return list, nil
+		},
+	}
+
+	r := NewReflector(lw, &v1.Pod{}, store, 0)
+	doneCh, stopCh := make(chan struct{}), make(chan struct{})
+	go func() {
+		defer close(doneCh)
+		r.Run(stopCh)
+	}()
+
+	// wait for the RV to sync to the version returned by the final list
+	err := wait.PollUntilContextTimeout(context.Background(), 100*time.Millisecond, wait.ForeverTestTimeout, true, func(ctx context.Context) (done bool, err error) {
+		if rv := r.LastSyncResourceVersion(); rv == lastExpectedRV {
+			return true, nil
+		}
+		return false, nil
+	})
+	if err != nil {
+		t.Fatalf("reflector never caught up with expected revision: %q, err: %v", lastExpectedRV, err)
+	}
+
+	if want, got := lastExpectedRV, r.LastSyncResourceVersion(); want != got {
+		t.Errorf("expected LastSyncResourceVersion to be %q, but got: %q", want, got)
+	}
+	if want, got := 2, int(replaceInvoked.Load()); want != got {
+		t.Errorf("expected store Delete hooks to be invoked %d times, but got: %d", want, got)
+	}
+	if want, got := len(pods), len(s.List()); want != got {
+		t.Errorf("expected the store to have %d objects, but got: %d", want, got)
+	}
+
+	close(stopCh)
+	select {
+	case <-doneCh:
+	case <-time.After(wait.ForeverTestTimeout):
+		t.Errorf("timed out waiting for Run to return")
+	}
+}
+
+type fakeStore struct {
+	Store
+	beforeReplace func(list []interface{}, s string)
+	afterReplace  func(rv string, err error)
+}
+
+func (f *fakeStore) Replace(list []interface{}, rv string) error {
+	f.beforeReplace(list, rv)
+	err := f.Store.Replace(list, rv)
+	f.afterReplace(rv, err)
+	return err
+}
+
 func BenchmarkExtractList(b *testing.B) {
 	_, _, podList := getPodListItems(0, fakeItemsNum)
 	_, _, configMapList := getConfigmapListItems(0, fakeItemsNum)