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,39 @@ 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("corrupt object has been deleted -%v", e.err.Error())
+}
+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 was corrupt, since we don't have the object data
+	// wrap the original error so we can include the partial object in the error chain
+	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,11 @@ func TestProgressNotify(t *testing.T) {
 	storagetesting.RunOptionalTestProgressNotify(ctx, t, store)
 }
 
+func TestWatchWithUnsafeDelete(t *testing.T) {
+	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{}, &storage.Preconditions{}, storage.ValidateAllObjectFunc, nil, storage.DeleteOptions{}); err == nil {
+		t.Fatalf("Expected normal Delete to fail")
+	}
+	if err := store.Delete(ctx, key, &example.Pod{}, &storage.Preconditions{}, 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 := "corrupt object has been deleted"; !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,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 	"math/rand"
+	"net/http"
 	"reflect"
 	goruntime "runtime"
 	"strconv"
@@ -1753,6 +1754,63 @@ func TestReflectorListExtract(t *testing.T) {
 	}
 }
 
+func TestReflectorWithUnsafeDelete(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",
+		}
+	}
+
+	list := mkList("1")
+	events := []watch.Event{
+		{Type: watch.Added, Object: mkPod("foo", "2")},
+		// the object gets corrupt, and it gets deleted
+		{Type: watch.Error, Object: makeStatus()},
+	}
+
+	s := NewFIFO(MetaNamespaceKeyFunc)
+	stopCh := make(chan struct{})
+	lw := &testLW{
+		WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
+			fw := watch.NewFake()
+			go func() {
+				for _, e := range events {
+					fw.Action(e.Type, e.Object)
+				}
+				// Because FakeWatcher doesn't buffer events, it's safe to
+				// close the stop channel immediately without missing events.
+				// But usually, the event producer would instead close the
+				// result channel, and wait for the consumer to stop the
+				// watcher, to avoid race conditions.
+				// TODO: Fix the FakeWatcher to separate watcher.Stop from close(resultCh)
+				close(stopCh)
+			}()
+			return fw, nil
+		},
+		ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
+			return list, nil
+		},
+	}
+
+	r := NewReflector(lw, &v1.Pod{}, s, 0)
+	if err := r.ListAndWatch(stopCh); err != nil {
+		t.Errorf("unexpected error from ListAndWatch: %v", err)
+	}
+}
+
 func BenchmarkExtractList(b *testing.B) {
 	_, _, podList := getPodListItems(0, fakeItemsNum)
 	_, _, configMapList := getConfigmapListItems(0, fakeItemsNum)