rebase: update K8s packages to v0.32.1

Update K8s packages in go.mod to v0.32.1

Signed-off-by: Praveen M <m.praveen@ibm.com>

vendor/k8s.io/apiserver/pkg/storage/etcd3/corrupt_obj_deleter.go

@@ -0,0 +1,270 @@
+/*
+Copyright 2024 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package etcd3
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"strings"
+
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	utilerrors "k8s.io/apimachinery/pkg/util/errors"
+	"k8s.io/apiserver/pkg/storage"
+	"k8s.io/apiserver/pkg/storage/value"
+	"k8s.io/klog/v2"
+)
+
+// NewStoreWithUnsafeCorruptObjectDeletion wraps the given store implementation
+// and adds support for unsafe deletion of corrupt objects
+func NewStoreWithUnsafeCorruptObjectDeletion(delegate storage.Interface, gr schema.GroupResource) storage.Interface {
+	return &corruptObjectDeleter{
+		Interface:     delegate,
+		groupResource: gr,
+	}
+}
+
+// WithCorruptObjErrorHandlingDecoder decorates the given decoder, it determines
+// if the error returned by the given decoder represents a corrupt object (the
+// object is undecodable), and then it wraps the error appropriately so the
+// unsafe deleter can determine if the object is a candidate for unsafe deletion
+func WithCorruptObjErrorHandlingDecoder(decoder Decoder) Decoder {
+	return &corruptObjErrorInterpretingDecoder{Decoder: decoder}
+}
+
+// WithCorruptObjErrorHandlingTransformer decorates the given decoder, it
+// determines if the error returned by the given transformer represents a
+// corrupt object (the data from the storage is untransformable), and then it
+// wraps the error appropriately so the unsafe deleter can determine
+// if the object is a candidate for unsafe deletion
+func WithCorruptObjErrorHandlingTransformer(transformer value.Transformer) value.Transformer {
+	return &corruptObjErrorInterpretingTransformer{Transformer: transformer}
+}
+
+// corruptObjErrAggregatorFactory returns an error aggregator that aggregates
+// corrupt object error(s) that the list operation encounters while
+// retrieving objects from the storage.
+// maxCount: it is the maximum number of error that will be aggregated
+func corruptObjErrAggregatorFactory(maxCount int) func() ListErrorAggregator {
+	if maxCount <= 0 {
+		return defaultListErrorAggregatorFactory
+	}
+	return func() ListErrorAggregator {
+		return &corruptObjErrAggregator{maxCount: maxCount}
+	}
+}
+
+var errTooMany = errors.New("too many errors, the list is truncated")
+
+// aggregate corrupt object errors from the LIST operation
+type corruptObjErrAggregator struct {
+	errs     []error
+	abortErr error
+	maxCount int
+}
+
+func (a *corruptObjErrAggregator) Aggregate(key string, err error) bool {
+	if len(a.errs) >= a.maxCount {
+		// add a sentinel error to indicate there are more
+		a.errs = append(a.errs, errTooMany)
+		return true
+	}
+	var corruptObjErr *corruptObjectError
+	if errors.As(err, &corruptObjErr) {
+		a.errs = append(a.errs, storage.NewCorruptObjError(key, corruptObjErr))
+		return false
+	}
+
+	// not a corrupt object error, the list operation should abort
+	a.abortErr = err
+	return true
+}
+
+func (a *corruptObjErrAggregator) Err() error {
+	switch {
+	case len(a.errs) == 0 && a.abortErr != nil:
+		return a.abortErr
+	case len(a.errs) > 0:
+		err := utilerrors.NewAggregate(a.errs)
+		return &aggregatedStorageError{errs: err, resourcePrefix: "list"}
+	default:
+		return nil
+	}
+}
+
+// corruptObjectDeleter facilitates unsafe deletion of corrupt objects for etcd
+type corruptObjectDeleter struct {
+	storage.Interface
+	groupResource schema.GroupResource
+}
+
+func (s *corruptObjectDeleter) Get(ctx context.Context, key string, opts storage.GetOptions, out runtime.Object) error {
+	if err := s.Interface.Get(ctx, key, opts, out); err != nil {
+		var corruptObjErr *corruptObjectError
+		if !errors.As(err, &corruptObjErr) {
+			// this error does not represent a corrupt object
+			return err
+		}
+		// the unsafe deleter at the registry layer will check whether
+		// the given err represents a corrupt object in order to
+		// initiate the unsafe deletion flow.
+		return storage.NewCorruptObjError(key, corruptObjErr)
+	}
+	return nil
+}
+
+func (s *corruptObjectDeleter) GetList(ctx context.Context, key string, opts storage.ListOptions, listObj runtime.Object) error {
+	err := s.Interface.GetList(ctx, key, opts, listObj)
+	if err == nil {
+		return nil
+	}
+
+	var aggregatedErr *aggregatedStorageError
+	if errors.As(err, &aggregatedErr) {
+		// we have aggregated a list of corrupt objects
+		klog.V(5).ErrorS(aggregatedErr, "corrupt objects")
+		return aggregatedErr.NewAPIStatusError(s.groupResource)
+	}
+	return err
+}
+
+// corruptObjErrorInterpretingDecoder wraps the error returned by the decorated decoder
+type corruptObjErrorInterpretingDecoder struct {
+	Decoder
+}
+
+func (d *corruptObjErrorInterpretingDecoder) Decode(value []byte, objPtr runtime.Object, rev int64) error {
+	// TODO: right now any error is deemed as undecodable, in
+	// the future, we can apply some filter, if need be.
+	if err := d.Decoder.Decode(value, objPtr, rev); err != nil {
+		return &corruptObjectError{err: err, errType: undecodable, revision: rev}
+	}
+	return nil
+}
+
+// decodeListItem decodes bytes value in array into object.
+func (d *corruptObjErrorInterpretingDecoder) DecodeListItem(ctx context.Context, data []byte, rev uint64, newItemFunc func() runtime.Object) (runtime.Object, error) {
+	// TODO: right now any error is deemed as undecodable, in
+	// the future, we can apply some filter, if need be.
+	obj, err := d.Decoder.DecodeListItem(ctx, data, rev, newItemFunc)
+	if err != nil {
+		err = &corruptObjectError{err: err, errType: undecodable, revision: int64(rev)}
+	}
+	return obj, err
+}
+
+// corruptObjErrorInterpretingTransformer wraps the error returned by the transformer
+type corruptObjErrorInterpretingTransformer struct {
+	value.Transformer
+}
+
+func (t *corruptObjErrorInterpretingTransformer) TransformFromStorage(ctx context.Context, data []byte, dataCtx value.Context) ([]byte, bool, error) {
+	// TODO: right now any error is deemed as undecodable, in the future, we
+	// can apply some filter, if need be. For example, any network error
+	out, stale, err := t.Transformer.TransformFromStorage(ctx, data, dataCtx)
+	if err != nil {
+		err = &corruptObjectError{err: err, errType: untransformable}
+	}
+	return out, stale, err
+}
+
+// corruptObjectError is used internally, only by the corrupt object
+// deleter, this error represents a corrup object:
+// a) the data from the storage failed to transform, or
+// b) the data failed to decode into an object
+// NOTE: this error does not have any information to identify the object
+// that is corrupt, for example the storage key associated with the object
+type corruptObjectError struct {
+	err      error
+	errType  int
+	revision int64
+}
+
+const (
+	untransformable int = iota + 1
+	undecodable
+)
+
+var typeToMessage = map[int]string{
+	untransformable: "data from the storage is not transformable",
+	undecodable:     "object not decodable",
+}
+
+func (e *corruptObjectError) Unwrap() error { return e.err }
+func (e *corruptObjectError) Error() string {
+	return fmt.Sprintf("%s revision=%d: %v", typeToMessage[e.errType], e.revision, e.err)
+}
+
+// aggregatedStorageError holds an aggregated list of storage.StorageError
+type aggregatedStorageError struct {
+	resourcePrefix string
+	errs           utilerrors.Aggregate
+}
+
+func (e *aggregatedStorageError) Error() string {
+	errs := e.errs.Errors()
+	var b strings.Builder
+	fmt.Fprintf(&b, "unable to transform or decode %d objects: {\n", len(errs))
+	for _, err := range errs {
+		fmt.Fprintf(&b, "\t%s\n", err.Error())
+	}
+	b.WriteString("}")
+	return b.String()
+}
+
+// NewAPIStatusError creates a new APIStatus object from the
+// aggregated list of StorageError
+func (e *aggregatedStorageError) NewAPIStatusError(qualifiedResource schema.GroupResource) *apierrors.StatusError {
+	var causes []metav1.StatusCause
+	for _, err := range e.errs.Errors() {
+		var storageErr *storage.StorageError
+		if errors.As(err, &storageErr) {
+			causes = append(causes, metav1.StatusCause{
+				Type:  metav1.CauseTypeUnexpectedServerResponse,
+				Field: storageErr.Key,
+				// TODO: do we need to expose the internal error message here?
+				Message: err.Error(),
+			})
+			continue
+		}
+		if errors.Is(err, errTooMany) {
+			causes = append(causes, metav1.StatusCause{
+				Type:    metav1.CauseTypeTooMany,
+				Message: errTooMany.Error(),
+			})
+		}
+	}
+
+	return &apierrors.StatusError{
+		ErrStatus: metav1.Status{
+			Status: metav1.StatusFailure,
+			Code:   http.StatusInternalServerError,
+			Reason: metav1.StatusReasonStoreReadError,
+			Details: &metav1.StatusDetails{
+				Group:  qualifiedResource.Group,
+				Kind:   qualifiedResource.Resource,
+				Name:   e.resourcePrefix,
+				Causes: causes,
+			},
+			Message: fmt.Sprintf("failed to read one or more %s from the storage", qualifiedResource.String()),
+		},
+	}
+}

vendor/k8s.io/apiserver/pkg/storage/etcd3/decoder.go

@@ -0,0 +1,94 @@
+/*
+Copyright 2024 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package etcd3
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"k8s.io/apimachinery/pkg/conversion"
+	"k8s.io/apimachinery/pkg/runtime"
+	endpointsrequest "k8s.io/apiserver/pkg/endpoints/request"
+	"k8s.io/apiserver/pkg/storage"
+
+	"k8s.io/klog/v2"
+)
+
+// NewDefaultDecoder returns the default decoder for etcd3 store
+func NewDefaultDecoder(codec runtime.Codec, versioner storage.Versioner) Decoder {
+	return &defaultDecoder{
+		codec:     codec,
+		versioner: versioner,
+	}
+}
+
+// Decoder is used by the etcd storage implementation to decode
+// transformed data from the storage into an object
+type Decoder interface {
+	// Decode decodes value of bytes into object. It will also
+	// set the object resource version to rev.
+	// On success, objPtr would be set to the object.
+	Decode(value []byte, objPtr runtime.Object, rev int64) error
+
+	// DecodeListItem decodes bytes value in array into object.
+	DecodeListItem(ctx context.Context, data []byte, rev uint64, newItemFunc func() runtime.Object) (runtime.Object, error)
+}
+
+var _ Decoder = &defaultDecoder{}
+
+type defaultDecoder struct {
+	codec     runtime.Codec
+	versioner storage.Versioner
+}
+
+// decode decodes value of bytes into object. It will also set the object resource version to rev.
+// On success, objPtr would be set to the object.
+func (d *defaultDecoder) Decode(value []byte, objPtr runtime.Object, rev int64) error {
+	if _, err := conversion.EnforcePtr(objPtr); err != nil {
+		// nolint:errorlint // this code was moved from store.go as is
+		return fmt.Errorf("unable to convert output object to pointer: %v", err)
+	}
+	_, _, err := d.codec.Decode(value, nil, objPtr)
+	if err != nil {
+		return err
+	}
+	// being unable to set the version does not prevent the object from being extracted
+	if err := d.versioner.UpdateObject(objPtr, uint64(rev)); err != nil {
+		klog.Errorf("failed to update object version: %v", err)
+	}
+	return nil
+}
+
+// decodeListItem decodes bytes value in array into object.
+func (d *defaultDecoder) DecodeListItem(ctx context.Context, data []byte, rev uint64, newItemFunc func() runtime.Object) (runtime.Object, error) {
+	startedAt := time.Now()
+	defer func() {
+		endpointsrequest.TrackDecodeLatency(ctx, time.Since(startedAt))
+	}()
+
+	obj, _, err := d.codec.Decode(data, nil, newItemFunc())
+	if err != nil {
+		return nil, err
+	}
+
+	if err := d.versioner.UpdateObject(obj, rev); err != nil {
+		klog.Errorf("failed to update object version: %v", err)
+	}
+
+	return obj, nil
+}

vendor/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
 }
 

vendor/k8s.io/apiserver/pkg/storage/etcd3/store.go

@@ -19,14 +19,15 @@ package etcd3
 import (
 	"bytes"
 	"context"
-	"errors"
 	"fmt"
 	"path"
 	"reflect"
 	"strings"
 	"time"
 
+	"go.etcd.io/etcd/api/v3/mvccpb"
 	clientv3 "go.etcd.io/etcd/client/v3"
+	"go.etcd.io/etcd/client/v3/kubernetes"
 	"go.opentelemetry.io/otel/attribute"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -38,7 +39,6 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/apiserver/pkg/audit"
-	endpointsrequest "k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/apiserver/pkg/features"
 	"k8s.io/apiserver/pkg/storage"
 	"k8s.io/apiserver/pkg/storage/etcd3/metrics"
@@ -73,7 +73,7 @@ func (d authenticatedDataString) AuthenticatedData() []byte {
 var _ value.Context = authenticatedDataString("")
 
 type store struct {
-	client              *clientv3.Client
+	client              *kubernetes.Client
 	codec               runtime.Codec
 	versioner           storage.Versioner
 	transformer         value.Transformer
@@ -82,6 +82,8 @@ type store struct {
 	groupResourceString string
 	watcher             *watcher
 	leaseManager        *leaseManager
+	decoder             Decoder
+	listErrAggrFactory  func() ListErrorAggregator
 }
 
 func (s *store) RequestWatchProgress(ctx context.Context) error {
@@ -98,13 +100,52 @@ type objState struct {
 	stale bool
 }
 
-// New returns an etcd3 implementation of storage.Interface.
-func New(c *clientv3.Client, codec runtime.Codec, newFunc, newListFunc func() runtime.Object, prefix, resourcePrefix string, groupResource schema.GroupResource, transformer value.Transformer, leaseManagerConfig LeaseManagerConfig) storage.Interface {
-	return newStore(c, codec, newFunc, newListFunc, prefix, resourcePrefix, groupResource, transformer, leaseManagerConfig)
+// ListErrorAggregator aggregates the error(s) that the LIST operation
+// encounters while retrieving object(s) from the storage
+type ListErrorAggregator interface {
+	// Aggregate aggregates the given error from list operation
+	// key: it identifies the given object in the storage.
+	// err: it represents the error the list operation encountered while
+	// retrieving the given object from the storage.
+	// done: true if the aggregation is done and the list operation should
+	// abort, otherwise the list operation will continue
+	Aggregate(key string, err error) bool
+
+	// Err returns the aggregated error
+	Err() error
 }
 
-func newStore(c *clientv3.Client, codec runtime.Codec, newFunc, newListFunc func() runtime.Object, prefix, resourcePrefix string, groupResource schema.GroupResource, transformer value.Transformer, leaseManagerConfig LeaseManagerConfig) *store {
-	versioner := storage.APIObjectVersioner{}
+// defaultListErrorAggregatorFactory returns the default list error
+// aggregator that maintains backward compatibility, which is abort
+// the list operation as soon as it encounters the first error
+func defaultListErrorAggregatorFactory() ListErrorAggregator { return &abortOnFirstError{} }
+
+// LIST aborts on the first error it encounters (backward compatible)
+type abortOnFirstError struct {
+	err error
+}
+
+func (a *abortOnFirstError) Aggregate(key string, err error) bool {
+	a.err = err
+	return true
+}
+func (a *abortOnFirstError) Err() error { return a.err }
+
+// New returns an etcd3 implementation of storage.Interface.
+func New(c *kubernetes.Client, codec runtime.Codec, newFunc, newListFunc func() runtime.Object, prefix, resourcePrefix string, groupResource schema.GroupResource, transformer value.Transformer, leaseManagerConfig LeaseManagerConfig, decoder Decoder, versioner storage.Versioner) storage.Interface {
+	if utilfeature.DefaultFeatureGate.Enabled(features.AllowUnsafeMalformedObjectDeletion) {
+		transformer = WithCorruptObjErrorHandlingTransformer(transformer)
+		decoder = WithCorruptObjErrorHandlingDecoder(decoder)
+	}
+	var store storage.Interface
+	store = newStore(c, codec, newFunc, newListFunc, prefix, resourcePrefix, groupResource, transformer, leaseManagerConfig, decoder, versioner)
+	if utilfeature.DefaultFeatureGate.Enabled(features.AllowUnsafeMalformedObjectDeletion) {
+		store = NewStoreWithUnsafeCorruptObjectDeletion(store, groupResource)
+	}
+	return store
+}
+
+func newStore(c *kubernetes.Client, codec runtime.Codec, newFunc, newListFunc func() runtime.Object, prefix, resourcePrefix string, groupResource schema.GroupResource, transformer value.Transformer, leaseManagerConfig LeaseManagerConfig, decoder Decoder, versioner storage.Versioner) *store {
 	// for compatibility with etcd2 impl.
 	// no-op for default prefix of '/registry'.
 	// keeps compatibility with etcd2 impl for custom prefixes that don't start with '/'
@@ -114,8 +155,13 @@ func newStore(c *clientv3.Client, codec runtime.Codec, newFunc, newListFunc func
 		pathPrefix += "/"
 	}
 
+	listErrAggrFactory := defaultListErrorAggregatorFactory
+	if utilfeature.DefaultFeatureGate.Enabled(features.AllowUnsafeMalformedObjectDeletion) {
+		listErrAggrFactory = corruptObjErrAggregatorFactory(100)
+	}
+
 	w := &watcher{
-		client:        c,
+		client:        c.Client,
 		codec:         codec,
 		newFunc:       newFunc,
 		groupResource: groupResource,
@@ -136,7 +182,9 @@ func newStore(c *clientv3.Client, codec runtime.Codec, newFunc, newListFunc func
 		groupResource:       groupResource,
 		groupResourceString: groupResource.String(),
 		watcher:             w,
-		leaseManager:        newDefaultLeaseManager(c, leaseManagerConfig),
+		leaseManager:        newDefaultLeaseManager(c.Client, leaseManagerConfig),
+		decoder:             decoder,
+		listErrAggrFactory:  listErrAggrFactory,
 	}
 
 	w.getCurrentStorageRV = func(ctx context.Context) (uint64, error) {
@@ -160,29 +208,28 @@ func (s *store) Get(ctx context.Context, key string, opts storage.GetOptions, ou
 		return err
 	}
 	startTime := time.Now()
-	getResp, err := s.client.KV.Get(ctx, preparedKey)
+	getResp, err := s.client.Kubernetes.Get(ctx, preparedKey, kubernetes.GetOptions{})
 	metrics.RecordEtcdRequest("get", s.groupResourceString, err, startTime)
 	if err != nil {
 		return err
 	}
-	if err = s.validateMinimumResourceVersion(opts.ResourceVersion, uint64(getResp.Header.Revision)); err != nil {
+	if err = s.validateMinimumResourceVersion(opts.ResourceVersion, uint64(getResp.Revision)); err != nil {
 		return err
 	}
 
-	if len(getResp.Kvs) == 0 {
+	if getResp.KV == nil {
 		if opts.IgnoreNotFound {
 			return runtime.SetZeroValue(out)
 		}
 		return storage.NewKeyNotFoundError(preparedKey, 0)
 	}
-	kv := getResp.Kvs[0]
 
-	data, _, err := s.transformer.TransformFromStorage(ctx, kv.Value, authenticatedDataString(preparedKey))
+	data, _, err := s.transformer.TransformFromStorage(ctx, getResp.KV.Value, authenticatedDataString(preparedKey))
 	if err != nil {
-		return storage.NewInternalError(err.Error())
+		return storage.NewInternalError(err)
 	}
 
-	err = decode(s.codec, s.versioner, data, out, kv.ModRevision)
+	err = s.decoder.Decode(data, out, getResp.KV.ModRevision)
 	if err != nil {
 		recordDecodeError(s.groupResourceString, preparedKey)
 		return err
@@ -217,24 +264,23 @@ func (s *store) Create(ctx context.Context, key string, obj, out runtime.Object,
 	}
 	span.AddEvent("Encode succeeded", attribute.Int("len", len(data)))
 
-	opts, err := s.ttlOpts(ctx, int64(ttl))
-	if err != nil {
-		return err
+	var lease clientv3.LeaseID
+	if ttl != 0 {
+		lease, err = s.leaseManager.GetLease(ctx, int64(ttl))
+		if err != nil {
+			return err
+		}
 	}
 
 	newData, err := s.transformer.TransformToStorage(ctx, data, authenticatedDataString(preparedKey))
 	if err != nil {
 		span.AddEvent("TransformToStorage failed", attribute.String("err", err.Error()))
-		return storage.NewInternalError(err.Error())
+		return storage.NewInternalError(err)
 	}
 	span.AddEvent("TransformToStorage succeeded")
 
 	startTime := time.Now()
-	txnResp, err := s.client.KV.Txn(ctx).If(
-		notFound(preparedKey),
-	).Then(
-		clientv3.OpPut(preparedKey, string(newData), opts...),
-	).Commit()
+	txnResp, err := s.client.Kubernetes.OptimisticPut(ctx, preparedKey, newData, 0, kubernetes.PutOptions{LeaseID: lease})
 	metrics.RecordEtcdRequest("create", s.groupResourceString, err, startTime)
 	if err != nil {
 		span.AddEvent("Txn call failed", attribute.String("err", err.Error()))
@@ -247,8 +293,7 @@ func (s *store) Create(ctx context.Context, key string, obj, out runtime.Object,
 	}
 
 	if out != nil {
-		putResp := txnResp.Responses[0].GetResponsePut()
-		err = decode(s.codec, s.versioner, data, out, putResp.Header.Revision)
+		err = s.decoder.Decode(data, out, txnResp.Revision)
 		if err != nil {
 			span.AddEvent("decode failed", attribute.Int("len", len(data)), attribute.String("err", err.Error()))
 			recordDecodeError(s.groupResourceString, preparedKey)
@@ -262,7 +307,7 @@ func (s *store) Create(ctx context.Context, key string, obj, out runtime.Object,
 // Delete implements storage.Interface.Delete.
 func (s *store) Delete(
 	ctx context.Context, key string, out runtime.Object, preconditions *storage.Preconditions,
-	validateDeletion storage.ValidateObjectFunc, cachedExistingObject runtime.Object) error {
+	validateDeletion storage.ValidateObjectFunc, cachedExistingObject runtime.Object, opts storage.DeleteOptions) error {
 	preparedKey, err := s.prepareKey(key)
 	if err != nil {
 		return err
@@ -271,13 +316,18 @@ func (s *store) Delete(
 	if err != nil {
 		return fmt.Errorf("unable to convert output object to pointer: %v", err)
 	}
-	return s.conditionalDelete(ctx, preparedKey, out, v, preconditions, validateDeletion, cachedExistingObject)
+
+	skipTransformDecode := false
+	if utilfeature.DefaultFeatureGate.Enabled(features.AllowUnsafeMalformedObjectDeletion) {
+		skipTransformDecode = opts.IgnoreStoreReadError
+	}
+	return s.conditionalDelete(ctx, preparedKey, out, v, preconditions, validateDeletion, cachedExistingObject, skipTransformDecode)
 }
 
 func (s *store) conditionalDelete(
 	ctx context.Context, key string, out runtime.Object, v reflect.Value, preconditions *storage.Preconditions,
-	validateDeletion storage.ValidateObjectFunc, cachedExistingObject runtime.Object) error {
-	getCurrentState := s.getCurrentState(ctx, key, v, false)
+	validateDeletion storage.ValidateObjectFunc, cachedExistingObject runtime.Object, skipTransformDecode bool) error {
+	getCurrentState := s.getCurrentState(ctx, key, v, false, skipTransformDecode)
 
 	var origState *objState
 	var err error
@@ -347,21 +397,16 @@ func (s *store) conditionalDelete(
 		}
 
 		startTime := time.Now()
-		txnResp, err := s.client.KV.Txn(ctx).If(
-			clientv3.Compare(clientv3.ModRevision(key), "=", origState.rev),
-		).Then(
-			clientv3.OpDelete(key),
-		).Else(
-			clientv3.OpGet(key),
-		).Commit()
+		txnResp, err := s.client.Kubernetes.OptimisticDelete(ctx, key, origState.rev, kubernetes.DeleteOptions{
+			GetOnFailure: true,
+		})
 		metrics.RecordEtcdRequest("delete", s.groupResourceString, err, startTime)
 		if err != nil {
 			return err
 		}
 		if !txnResp.Succeeded {
-			getResp := (*clientv3.GetResponse)(txnResp.Responses[0].GetResponseRange())
 			klog.V(4).Infof("deletion of %s failed because of a conflict, going to retry", key)
-			origState, err = s.getState(ctx, getResp, key, v, false)
+			origState, err = s.getState(ctx, txnResp.KV, key, v, false, skipTransformDecode)
 			if err != nil {
 				return err
 			}
@@ -369,17 +414,12 @@ func (s *store) conditionalDelete(
 			continue
 		}
 
-		if len(txnResp.Responses) == 0 || txnResp.Responses[0].GetResponseDeleteRange() == nil {
-			return errors.New(fmt.Sprintf("invalid DeleteRange response: %v", txnResp.Responses))
-		}
-		deleteResp := txnResp.Responses[0].GetResponseDeleteRange()
-		if deleteResp.Header == nil {
-			return errors.New("invalid DeleteRange response - nil header")
-		}
-		err = decode(s.codec, s.versioner, origState.data, out, deleteResp.Header.Revision)
-		if err != nil {
-			recordDecodeError(s.groupResourceString, key)
-			return err
+		if !skipTransformDecode {
+			err = s.decoder.Decode(origState.data, out, txnResp.Revision)
+			if err != nil {
+				recordDecodeError(s.groupResourceString, key)
+				return err
+			}
 		}
 		return nil
 	}
@@ -405,7 +445,8 @@ func (s *store) GuaranteedUpdate(
 		return fmt.Errorf("unable to convert output object to pointer: %v", err)
 	}
 
-	getCurrentState := s.getCurrentState(ctx, preparedKey, v, ignoreNotFound)
+	skipTransformDecode := false
+	getCurrentState := s.getCurrentState(ctx, preparedKey, v, ignoreNotFound, skipTransformDecode)
 
 	var origState *objState
 	var origStateIsCurrent bool
@@ -491,7 +532,7 @@ func (s *store) GuaranteedUpdate(
 			}
 			// recheck that the data from etcd is not stale before short-circuiting a write
 			if !origState.stale {
-				err = decode(s.codec, s.versioner, origState.data, destination, origState.rev)
+				err = s.decoder.Decode(origState.data, destination, origState.rev)
 				if err != nil {
 					recordDecodeError(s.groupResourceString, preparedKey)
 					return err
@@ -503,24 +544,25 @@ func (s *store) GuaranteedUpdate(
 		newData, err := s.transformer.TransformToStorage(ctx, data, transformContext)
 		if err != nil {
 			span.AddEvent("TransformToStorage failed", attribute.String("err", err.Error()))
-			return storage.NewInternalError(err.Error())
+			return storage.NewInternalError(err)
 		}
 		span.AddEvent("TransformToStorage succeeded")
 
-		opts, err := s.ttlOpts(ctx, int64(ttl))
-		if err != nil {
-			return err
+		var lease clientv3.LeaseID
+		if ttl != 0 {
+			lease, err = s.leaseManager.GetLease(ctx, int64(ttl))
+			if err != nil {
+				return err
+			}
 		}
 		span.AddEvent("Transaction prepared")
 
 		startTime := time.Now()
-		txnResp, err := s.client.KV.Txn(ctx).If(
-			clientv3.Compare(clientv3.ModRevision(preparedKey), "=", origState.rev),
-		).Then(
-			clientv3.OpPut(preparedKey, string(newData), opts...),
-		).Else(
-			clientv3.OpGet(preparedKey),
-		).Commit()
+
+		txnResp, err := s.client.Kubernetes.OptimisticPut(ctx, preparedKey, newData, origState.rev, kubernetes.PutOptions{
+			GetOnFailure: true,
+			LeaseID:      lease,
+		})
 		metrics.RecordEtcdRequest("update", s.groupResourceString, err, startTime)
 		if err != nil {
 			span.AddEvent("Txn call failed", attribute.String("err", err.Error()))
@@ -529,9 +571,8 @@ func (s *store) GuaranteedUpdate(
 		span.AddEvent("Txn call completed")
 		span.AddEvent("Transaction committed")
 		if !txnResp.Succeeded {
-			getResp := (*clientv3.GetResponse)(txnResp.Responses[0].GetResponseRange())
 			klog.V(4).Infof("GuaranteedUpdate of %s failed because of a conflict, going to retry", preparedKey)
-			origState, err = s.getState(ctx, getResp, preparedKey, v, ignoreNotFound)
+			origState, err = s.getState(ctx, txnResp.KV, preparedKey, v, ignoreNotFound, skipTransformDecode)
 			if err != nil {
 				return err
 			}
@@ -539,9 +580,8 @@ func (s *store) GuaranteedUpdate(
 			origStateIsCurrent = true
 			continue
 		}
-		putResp := txnResp.Responses[0].GetResponsePut()
 
-		err = decode(s.codec, s.versioner, data, destination, putResp.Header.Revision)
+		err = s.decoder.Decode(data, destination, txnResp.Revision)
 		if err != nil {
 			span.AddEvent("decode failed", attribute.Int("len", len(data)), attribute.String("err", err.Error()))
 			recordDecodeError(s.groupResourceString, preparedKey)
@@ -583,12 +623,12 @@ func (s *store) Count(key string) (int64, error) {
 	}
 
 	startTime := time.Now()
-	getResp, err := s.client.KV.Get(context.Background(), preparedKey, clientv3.WithRange(clientv3.GetPrefixRangeEnd(preparedKey)), clientv3.WithCountOnly())
+	count, err := s.client.Kubernetes.Count(context.Background(), preparedKey, kubernetes.CountOptions{})
 	metrics.RecordEtcdRequest("listWithCount", preparedKey, err, startTime)
 	if err != nil {
 		return 0, err
 	}
-	return getResp.Count, nil
+	return count, nil
 }
 
 // ReadinessCheck implements storage.Interface.
@@ -639,7 +679,7 @@ func (s *store) resolveGetListRev(continueKey string, continueRV int64, opts sto
 
 // GetList implements storage.Interface.
 func (s *store) GetList(ctx context.Context, key string, opts storage.ListOptions, listObj runtime.Object) error {
-	preparedKey, err := s.prepareKey(key)
+	keyPrefix, err := s.prepareKey(key)
 	if err != nil {
 		return err
 	}
@@ -664,27 +704,13 @@ func (s *store) GetList(ctx context.Context, key string, opts storage.ListOption
 	// get children "directories". e.g. if we have key "/a", "/a/b", "/ab", getting keys
 	// with prefix "/a" will return all three, while with prefix "/a/" will return only
 	// "/a/b" which is the correct answer.
-	if opts.Recursive && !strings.HasSuffix(preparedKey, "/") {
-		preparedKey += "/"
+	if opts.Recursive && !strings.HasSuffix(keyPrefix, "/") {
+		keyPrefix += "/"
 	}
-	keyPrefix := preparedKey
 
 	// set the appropriate clientv3 options to filter the returned data set
-	var limitOption *clientv3.OpOption
 	limit := opts.Predicate.Limit
-	var paging bool
-	options := make([]clientv3.OpOption, 0, 4)
-	if opts.Predicate.Limit > 0 {
-		paging = true
-		options = append(options, clientv3.WithLimit(limit))
-		limitOption = &options[len(options)-1]
-	}
-
-	if opts.Recursive {
-		rangeEnd := clientv3.GetPrefixRangeEnd(keyPrefix)
-		options = append(options, clientv3.WithRange(rangeEnd))
-	}
-
+	paging := opts.Predicate.Limit > 0
 	newItemFunc := getNewItemFunc(listObj, v)
 
 	var continueRV, withRev int64
@@ -694,20 +720,15 @@ func (s *store) GetList(ctx context.Context, key string, opts storage.ListOption
 		if err != nil {
 			return apierrors.NewBadRequest(fmt.Sprintf("invalid continue token: %v", err))
 		}
-		preparedKey = continueKey
 	}
 	if withRev, err = s.resolveGetListRev(continueKey, continueRV, opts); err != nil {
 		return err
 	}
 
-	if withRev != 0 {
-		options = append(options, clientv3.WithRev(withRev))
-	}
-
 	// loop until we have filled the requested limit from etcd or there are no more results
 	var lastKey []byte
 	var hasMore bool
-	var getResp *clientv3.GetResponse
+	var getResp kubernetes.ListResponse
 	var numFetched int
 	var numEvald int
 	// Because these metrics are for understanding the costs of handling LIST requests,
@@ -722,26 +743,30 @@ func (s *store) GetList(ctx context.Context, key string, opts storage.ListOption
 		metricsOp = "list"
 	}
 
+	aggregator := s.listErrAggrFactory()
 	for {
 		startTime := time.Now()
-		getResp, err = s.client.KV.Get(ctx, preparedKey, options...)
+		getResp, err = s.getList(ctx, keyPrefix, opts.Recursive, kubernetes.ListOptions{
+			Revision: withRev,
+			Limit:    limit,
+			Continue: continueKey,
+		})
 		metrics.RecordEtcdRequest(metricsOp, s.groupResourceString, err, startTime)
 		if err != nil {
 			return interpretListError(err, len(opts.Predicate.Continue) > 0, continueKey, keyPrefix)
 		}
 		numFetched += len(getResp.Kvs)
-		if err = s.validateMinimumResourceVersion(opts.ResourceVersion, uint64(getResp.Header.Revision)); err != nil {
+		if err = s.validateMinimumResourceVersion(opts.ResourceVersion, uint64(getResp.Revision)); err != nil {
 			return err
 		}
-		hasMore = getResp.More
+		hasMore = int64(len(getResp.Kvs)) < getResp.Count
 
-		if len(getResp.Kvs) == 0 && getResp.More {
+		if len(getResp.Kvs) == 0 && hasMore {
 			return fmt.Errorf("no results were found, but etcd indicated there were more values remaining")
 		}
 		// indicate to the client which resource version was returned, and use the same resource version for subsequent requests.
 		if withRev == 0 {
-			withRev = getResp.Header.Revision
-			options = append(options, clientv3.WithRev(withRev))
+			withRev = getResp.Revision
 		}
 
 		// avoid small allocations for the result slice, since this can be called in many
@@ -762,7 +787,10 @@ func (s *store) GetList(ctx context.Context, key string, opts storage.ListOption
 
 			data, _, err := s.transformer.TransformFromStorage(ctx, kv.Value, authenticatedDataString(kv.Key))
 			if err != nil {
-				return storage.NewInternalErrorf("unable to transform key %q: %v", kv.Key, err)
+				if done := aggregator.Aggregate(string(kv.Key), storage.NewInternalError(fmt.Errorf("unable to transform key %q: %w", kv.Key, err))); done {
+					return aggregator.Err()
+				}
+				continue
 			}
 
 			// Check if the request has already timed out before decode object
@@ -773,10 +801,13 @@ func (s *store) GetList(ctx context.Context, key string, opts storage.ListOption
 			default:
 			}
 
-			obj, err := decodeListItem(ctx, data, uint64(kv.ModRevision), s.codec, s.versioner, newItemFunc)
+			obj, err := s.decoder.DecodeListItem(ctx, data, uint64(kv.ModRevision), newItemFunc)
 			if err != nil {
 				recordDecodeError(s.groupResourceString, string(kv.Key))
-				return err
+				if done := aggregator.Aggregate(string(kv.Key), err); done {
+					return aggregator.Err()
+				}
+				continue
 			}
 
 			// being unable to set the version does not prevent the object from being extracted
@@ -789,6 +820,7 @@ func (s *store) GetList(ctx context.Context, key string, opts storage.ListOption
 			// free kv early. Long lists can take O(seconds) to decode.
 			getResp.Kvs[i] = nil
 		}
+		continueKey = string(lastKey) + "\x00"
 
 		// no more results remain or we didn't request paging
 		if !hasMore || !paging {
@@ -806,9 +838,11 @@ func (s *store) GetList(ctx context.Context, key string, opts storage.ListOption
 			if limit > maxLimit {
 				limit = maxLimit
 			}
-			*limitOption = clientv3.WithLimit(limit)
 		}
-		preparedKey = string(lastKey) + "\x00"
+	}
+
+	if err := aggregator.Err(); err != nil {
+		return err
 	}
 
 	if v.IsNil() {
@@ -823,6 +857,26 @@ func (s *store) GetList(ctx context.Context, key string, opts storage.ListOption
 	return s.versioner.UpdateList(listObj, uint64(withRev), continueValue, remainingItemCount)
 }
 
+func (s *store) getList(ctx context.Context, keyPrefix string, recursive bool, options kubernetes.ListOptions) (kubernetes.ListResponse, error) {
+	if recursive {
+		return s.client.Kubernetes.List(ctx, keyPrefix, options)
+	}
+	getResp, err := s.client.Kubernetes.Get(ctx, keyPrefix, kubernetes.GetOptions{
+		Revision: options.Revision,
+	})
+	var resp kubernetes.ListResponse
+	if getResp.KV != nil {
+		resp.Kvs = []*mvccpb.KeyValue{getResp.KV}
+		resp.Count = 1
+		resp.Revision = getResp.Revision
+	} else {
+		resp.Kvs = []*mvccpb.KeyValue{}
+		resp.Count = 0
+		resp.Revision = getResp.Revision
+	}
+	return resp, err
+}
+
 // growSlice takes a slice value and grows its capacity up
 // to the maximum of the passed sizes or maxCapacity, whichever
 // is smaller. Above maxCapacity decisions about allocation are left
@@ -878,19 +932,25 @@ func (s *store) watchContext(ctx context.Context) context.Context {
 	return clientv3.WithRequireLeader(ctx)
 }
 
-func (s *store) getCurrentState(ctx context.Context, key string, v reflect.Value, ignoreNotFound bool) func() (*objState, error) {
+func (s *store) getCurrentState(ctx context.Context, key string, v reflect.Value, ignoreNotFound bool, skipTransformDecode bool) func() (*objState, error) {
 	return func() (*objState, error) {
 		startTime := time.Now()
-		getResp, err := s.client.KV.Get(ctx, key)
+		getResp, err := s.client.Kubernetes.Get(ctx, key, kubernetes.GetOptions{})
 		metrics.RecordEtcdRequest("get", s.groupResourceString, err, startTime)
 		if err != nil {
 			return nil, err
 		}
-		return s.getState(ctx, getResp, key, v, ignoreNotFound)
+		return s.getState(ctx, getResp.KV, key, v, ignoreNotFound, skipTransformDecode)
 	}
 }
 
-func (s *store) getState(ctx context.Context, getResp *clientv3.GetResponse, key string, v reflect.Value, ignoreNotFound bool) (*objState, error) {
+// getState constructs a new objState from the given response from the storage.
+// skipTransformDecode: if true, the function will neither transform the data
+// from the storage nor decode it into an object; otherwise, data from the
+// storage will be transformed and decoded.
+// NOTE: when skipTransformDecode is true, the 'data', and the 'obj' fields
+// of the objState will be nil, and 'stale' will be set to true.
+func (s *store) getState(ctx context.Context, kv *mvccpb.KeyValue, key string, v reflect.Value, ignoreNotFound bool, skipTransformDecode bool) (*objState, error) {
 	state := &objState{
 		meta: &storage.ResponseMeta{},
 	}
@@ -901,7 +961,7 @@ func (s *store) getState(ctx context.Context, getResp *clientv3.GetResponse, key
 		state.obj = reflect.New(v.Type()).Interface().(runtime.Object)
 	}
 
-	if len(getResp.Kvs) == 0 {
+	if kv == nil {
 		if !ignoreNotFound {
 			return nil, storage.NewKeyNotFoundError(key, 0)
 		}
@@ -909,15 +969,25 @@ func (s *store) getState(ctx context.Context, getResp *clientv3.GetResponse, key
 			return nil, err
 		}
 	} else {
-		data, stale, err := s.transformer.TransformFromStorage(ctx, getResp.Kvs[0].Value, authenticatedDataString(key))
-		if err != nil {
-			return nil, storage.NewInternalError(err.Error())
-		}
-		state.rev = getResp.Kvs[0].ModRevision
+		state.rev = kv.ModRevision
 		state.meta.ResourceVersion = uint64(state.rev)
+
+		if skipTransformDecode {
+			// be explicit that we don't have the object
+			state.obj = nil
+			state.stale = true // this seems a more sane value here
+			return state, nil
+		}
+
+		data, stale, err := s.transformer.TransformFromStorage(ctx, kv.Value, authenticatedDataString(key))
+		if err != nil {
+			return nil, storage.NewInternalError(err)
+		}
+
 		state.data = data
 		state.stale = stale
-		if err := decode(s.codec, s.versioner, state.data, state.obj, state.rev); err != nil {
+
+		if err := s.decoder.Decode(state.data, state.obj, state.rev); err != nil {
 			recordDecodeError(s.groupResourceString, key)
 			return nil, err
 		}
@@ -969,19 +1039,6 @@ func (s *store) updateState(st *objState, userUpdate storage.UpdateFunc) (runtim
 	return ret, ttl, nil
 }
 
-// ttlOpts returns client options based on given ttl.
-// ttl: if ttl is non-zero, it will attach the key to a lease with ttl of roughly the same length
-func (s *store) ttlOpts(ctx context.Context, ttl int64) ([]clientv3.OpOption, error) {
-	if ttl == 0 {
-		return nil, nil
-	}
-	id, err := s.leaseManager.GetLease(ctx, ttl)
-	if err != nil {
-		return nil, err
-	}
-	return []clientv3.OpOption{clientv3.WithLease(id)}, nil
-}
-
 // validateMinimumResourceVersion returns a 'too large resource' version error when the provided minimumResourceVersion is
 // greater than the most recent actualRevision available from storage.
 func (s *store) validateMinimumResourceVersion(minimumResourceVersion string, actualRevision uint64) error {
@@ -1024,52 +1081,12 @@ func (s *store) prepareKey(key string) (string, error) {
 	return s.pathPrefix + key[startIndex:], nil
 }
 
-// decode decodes value of bytes into object. It will also set the object resource version to rev.
-// On success, objPtr would be set to the object.
-func decode(codec runtime.Codec, versioner storage.Versioner, value []byte, objPtr runtime.Object, rev int64) error {
-	if _, err := conversion.EnforcePtr(objPtr); err != nil {
-		return fmt.Errorf("unable to convert output object to pointer: %v", err)
-	}
-	_, _, err := codec.Decode(value, nil, objPtr)
-	if err != nil {
-		return err
-	}
-	// being unable to set the version does not prevent the object from being extracted
-	if err := versioner.UpdateObject(objPtr, uint64(rev)); err != nil {
-		klog.Errorf("failed to update object version: %v", err)
-	}
-	return nil
-}
-
-// decodeListItem decodes bytes value in array into object.
-func decodeListItem(ctx context.Context, data []byte, rev uint64, codec runtime.Codec, versioner storage.Versioner, newItemFunc func() runtime.Object) (runtime.Object, error) {
-	startedAt := time.Now()
-	defer func() {
-		endpointsrequest.TrackDecodeLatency(ctx, time.Since(startedAt))
-	}()
-
-	obj, _, err := codec.Decode(data, nil, newItemFunc())
-	if err != nil {
-		return nil, err
-	}
-
-	if err := versioner.UpdateObject(obj, rev); err != nil {
-		klog.Errorf("failed to update object version: %v", err)
-	}
-
-	return obj, nil
-}
-
 // recordDecodeError record decode error split by object type.
 func recordDecodeError(resource string, key string) {
 	metrics.RecordDecodeError(resource)
 	klog.V(4).Infof("Decoding %s \"%s\" failed", resource, key)
 }
 
-func notFound(key string) clientv3.Cmp {
-	return clientv3.Compare(clientv3.ModRevision(key), "=", 0)
-}
-
 // getTypeName returns type name of an object for reporting purposes.
 func getTypeName(obj interface{}) string {
 	return reflect.TypeOf(obj).String()

vendor/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 {