refactor: storage.InternalError should retain the inner error

This is so we can do proper error handling, at the same time
we want to maintain backward compatibility

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

@@ -172,24 +172,26 @@ func NewInvalidError(errors field.ErrorList) InvalidError {
 // not from the underlying storage backend (e.g., etcd).
 type InternalError struct {
 	Reason string
+
+	// retain the inner error to maintain the error tree, so as to enable us
+	// to do proper error checking, but we also need to be backward compatible.
+	err error
 }
 
 func (e InternalError) Error() string {
 	return e.Reason
 }
 
+func (e InternalError) Unwrap() error { return e.err }
+
 // IsInternalError returns true if and only if err is an InternalError.
 func IsInternalError(err error) bool {
 	_, ok := err.(InternalError)
 	return ok
 }
 
-func NewInternalError(reason string) InternalError {
+func NewInternalError(err error) InternalError {
-	return InternalError{reason}
+	return InternalError{Reason: err.Error(), err: err}
-}
-
-func NewInternalErrorf(format string, a ...interface{}) InternalError {
-	return InternalError{fmt.Sprintf(format, a...)}
 }
 
 var tooLargeResourceVersionCauseMsg = "Too large resource version"

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

@@ -179,7 +179,7 @@ func (s *store) Get(ctx context.Context, key string, opts storage.GetOptions, ou
 
 	data, _, err := s.transformer.TransformFromStorage(ctx, 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)
@@ -225,7 +225,7 @@ func (s *store) Create(ctx context.Context, key string, obj, out runtime.Object,
 	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")
 
@@ -508,7 +508,7 @@ 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")
 
@@ -768,7 +768,7 @@ 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)
+				return storage.NewInternalError(fmt.Errorf("unable to transform key %q: %w", kv.Key, err))
 			}
 
 			// Check if the request has already timed out before decode object
@@ -933,7 +933,7 @@ func (s *store) getState(ctx context.Context, getResp *clientv3.GetResponse, key
 
 		data, stale, err := s.transformer.TransformFromStorage(ctx, getResp.Kvs[0].Value, authenticatedDataString(key))
 		if err != nil {
-			return nil, storage.NewInternalError(err.Error())
+			return nil, storage.NewInternalError(err)
 		}
 
 		state.data = data

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

@@ -139,11 +139,11 @@ func (p *Preconditions) Check(key string, obj runtime.Object) error {
 	}
 	objMeta, err := meta.Accessor(obj)
 	if err != nil {
-		return NewInternalErrorf(
+		return NewInternalError(
-			"can't enforce preconditions %v on un-introspectable object %v, got error: %v",
+			fmt.Errorf("can't enforce preconditions %v on un-introspectable object %v, got error: %w",
-			*p,
+				*p,
-			obj,
+				obj,
-			err)
+				err))
 	}
 	if p.UID != nil && *p.UID != objMeta.GetUID() {
 		err := fmt.Sprintf(