api: add a new field to meta/v1 DeleteOptions

- add a new boolean field
  IgnoreStoreReadErrorWithClusterBreakingPotential to meta/v1 DeleteOptions

- add validation for the new delete option
add validation for the new field in the delete options
ignoreStoreReadErrorWithClusterBreakingPotential

- prevent the pod eviction handler from issuing an unsafe pod delete
prevent the pod eviction handler from enabling the
'ignoreStoreReadErrorWithClusterBreakingPotential' delete option

pkg/registry/core/pod/storage/eviction.go

@@ -30,6 +30,7 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/validation/field"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apiserver/pkg/registry/rest"
 	"k8s.io/apiserver/pkg/util/dryrun"
@@ -39,6 +40,8 @@ import (
 	podutil "k8s.io/kubernetes/pkg/api/pod"
 	api "k8s.io/kubernetes/pkg/apis/core"
 	"k8s.io/kubernetes/pkg/apis/policy"
+
+	"k8s.io/utils/ptr"
 )
 
 const (
@@ -133,6 +136,13 @@ func (r *EvictionREST) Create(ctx context.Context, name string, obj runtime.Obje
 		return nil, errors.NewBadRequest("name in URL does not match name in Eviction object")
 	}
 
+	if eviction.DeleteOptions != nil && ptr.Deref(eviction.DeleteOptions.IgnoreStoreReadErrorWithClusterBreakingPotential, false) {
+		errs := field.ErrorList{
+			field.Invalid(field.NewPath("deleteOptions").Child("ignoreStoreReadErrorWithClusterBreakingPotential"), true, "can not be set for pod eviction, try after removing the option"),
+		}
+		return nil, errors.NewInvalid(v1Eviction.GroupKind(), name, errs)
+	}
+
 	originalDeleteOptions, err := propagateDryRun(eviction, options)
 	if err != nil {
 		return nil, err

pkg/registry/core/pod/storage/eviction_test.go

@@ -24,6 +24,8 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/google/go-cmp/cmp"
+
 	policyv1 "k8s.io/api/policy/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
@@ -31,6 +33,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/validation/field"
 	"k8s.io/apimachinery/pkg/watch"
 	genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/apiserver/pkg/registry/rest"
@@ -38,6 +41,8 @@ import (
 	podapi "k8s.io/kubernetes/pkg/api/pod"
 	api "k8s.io/kubernetes/pkg/apis/core"
 	"k8s.io/kubernetes/pkg/apis/policy"
+
+	"k8s.io/utils/ptr"
 )
 
 func TestEviction(t *testing.T) {
@@ -581,30 +586,31 @@ func TestEvictionIgnorePDB(t *testing.T) {
 	}
 }
 
-func TestEvictionDryRun(t *testing.T) {
+func TestEvictionWithDeleteOptions(t *testing.T) {
 	testcases := []struct {
 		name            string
 		evictionOptions *metav1.DeleteOptions
 		requestOptions  *metav1.CreateOptions
 		pdbs            []runtime.Object
+		err             error
 	}{
 		{
-			name:            "just request-options",
+			name:            "dry run - just request-options",
 			requestOptions:  &metav1.CreateOptions{DryRun: []string{"All"}},
 			evictionOptions: &metav1.DeleteOptions{},
 		},
 		{
-			name:            "just eviction-options",
+			name:            "dry run - just eviction-options",
 			requestOptions:  &metav1.CreateOptions{},
 			evictionOptions: &metav1.DeleteOptions{DryRun: []string{"All"}},
 		},
 		{
-			name:            "both options",
+			name:            "dry run - both options",
 			evictionOptions: &metav1.DeleteOptions{DryRun: []string{"All"}},
 			requestOptions:  &metav1.CreateOptions{DryRun: []string{"All"}},
 		},
 		{
-			name:            "with pdbs",
+			name:            "dry run - with pdbs",
 			evictionOptions: &metav1.DeleteOptions{DryRun: []string{"All"}},
 			requestOptions:  &metav1.CreateOptions{DryRun: []string{"All"}},
 			pdbs: []runtime.Object{&policyv1.PodDisruptionBudget{
@@ -613,6 +619,17 @@ func TestEvictionDryRun(t *testing.T) {
 				Status:     policyv1.PodDisruptionBudgetStatus{DisruptionsAllowed: 1},
 			}},
 		},
+		{
+			name:           "ignoreStoreReadErrorWithClusterBreakingPotential is set, invalid error expected",
+			requestOptions: &metav1.CreateOptions{},
+			evictionOptions: &metav1.DeleteOptions{
+				IgnoreStoreReadErrorWithClusterBreakingPotential: ptr.To[bool](true),
+			},
+			err: apierrors.NewInvalid(v1Eviction.GroupKind(), "foo", field.ErrorList{
+				field.Invalid(field.NewPath("deleteOptions").Child("ignoreStoreReadErrorWithClusterBreakingPotential"),
+					true, "can not be set for pod eviction, try after removing the option"),
+			}),
+		},
 	}
 
 	for _, tc := range testcases {
@@ -633,8 +650,8 @@ func TestEvictionDryRun(t *testing.T) {
 			evictionRest := newEvictionStorage(storage.Store, client.PolicyV1())
 			eviction := &policy.Eviction{ObjectMeta: metav1.ObjectMeta{Name: "foo", Namespace: "default"}, DeleteOptions: tc.evictionOptions}
 			_, err := evictionRest.Create(testContext, pod.Name, eviction, nil, tc.requestOptions)
-			if err != nil {
-				t.Fatalf("Failed to run eviction: %v", err)
+			if !cmp.Equal(tc.err, err) {
+				t.Errorf("expected error to match, diff: %s", cmp.Diff(tc.err, err))
 			}
 		})
 	}

staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/types.go

@@ -560,6 +560,21 @@ type DeleteOptions struct {
 	// +optional
 	// +listType=atomic
 	DryRun []string `json:"dryRun,omitempty" protobuf:"bytes,5,rep,name=dryRun"`
+
+	// if set to true, it will trigger an unsafe deletion of the resource in
+	// case the normal deletion flow fails with a corrupt object error.
+	// A resource is considered corrupt if it can not be retrieved from
+	// the underlying storage successfully because of a) its data can
+	// not be transformed e.g. decryption failure, or b) it fails
+	// to decode into an object.
+	// NOTE: unsafe deletion ignores finalizer constraints, skips
+	// precondition checks, and removes the object from the storage.
+	// WARNING: This may potentially break the cluster if the workload
+	// associated with the resource being unsafe-deleted relies on normal
+	// deletion flow. Use only if you REALLY know what you are doing.
+	// The default value is false, and the user must opt in to enable it
+	// +optional
+	IgnoreStoreReadErrorWithClusterBreakingPotential *bool `json:"ignoreStoreReadErrorWithClusterBreakingPotential,omitempty" protobuf:"varint,6,opt,name=ignoreStoreReadErrorWithClusterBreakingPotential"`
 }
 
 const (

staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/validation/validation.go

@@ -26,6 +26,8 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/validation"
 	"k8s.io/apimachinery/pkg/util/validation/field"
+
+	"k8s.io/utils/ptr"
 )
 
 // LabelSelectorValidationOptions is a struct that can be passed to ValidateLabelSelector to record the validate options
@@ -165,6 +167,7 @@ func ValidateDeleteOptions(options *metav1.DeleteOptions) field.ErrorList {
 		allErrs = append(allErrs, field.NotSupported(field.NewPath("propagationPolicy"), options.PropagationPolicy, []string{string(metav1.DeletePropagationForeground), string(metav1.DeletePropagationBackground), string(metav1.DeletePropagationOrphan), "nil"}))
 	}
 	allErrs = append(allErrs, ValidateDryRun(field.NewPath("dryRun"), options.DryRun)...)
+	allErrs = append(allErrs, ValidateIgnoreStoreReadError(field.NewPath("ignoreStoreReadErrorWithClusterBreakingPotential"), options)...)
 	return allErrs
 }
 
@@ -358,3 +361,31 @@ func isValidConditionReason(value string) []string {
 	}
 	return nil
 }
+
+// ValidateIgnoreStoreReadError validates that delete options are valid when
+// ignoreStoreReadErrorWithClusterBreakingPotential is enabled
+func ValidateIgnoreStoreReadError(fldPath *field.Path, options *metav1.DeleteOptions) field.ErrorList {
+	allErrs := field.ErrorList{}
+	if enabled := ptr.Deref[bool](options.IgnoreStoreReadErrorWithClusterBreakingPotential, false); !enabled {
+		return allErrs
+	}
+
+	if len(options.DryRun) > 0 {
+		allErrs = append(allErrs, field.Invalid(fldPath, true, "cannot be set together with .dryRun"))
+	}
+	if options.PropagationPolicy != nil {
+		allErrs = append(allErrs, field.Invalid(fldPath, true, "cannot be set together with .propagationPolicy"))
+	}
+	//nolint:staticcheck // Keep validation for deprecated OrphanDependents option until it's being removed
+	if options.OrphanDependents != nil {
+		allErrs = append(allErrs, field.Invalid(fldPath, true, "cannot be set together with .orphanDependents"))
+	}
+	if options.GracePeriodSeconds != nil {
+		allErrs = append(allErrs, field.Invalid(fldPath, true, "cannot be set together with .gracePeriodSeconds"))
+	}
+	if options.Preconditions != nil {
+		allErrs = append(allErrs, field.Invalid(fldPath, true, "cannot be set together with .preconditions"))
+	}
+
+	return allErrs
+}

staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/validation/validation_test.go

@@ -21,9 +21,13 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/google/go-cmp/cmp"
+
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/validation/field"
+
+	"k8s.io/utils/ptr"
 )
 
 func TestValidateLabels(t *testing.T) {
@@ -132,6 +136,96 @@ func boolPtr(b bool) *bool {
 	return &b
 }
 
+func TestValidateDeleteOptionsWithIgnoreStoreReadError(t *testing.T) {
+	fieldPath := field.NewPath("ignoreStoreReadErrorWithClusterBreakingPotential")
+	tests := []struct {
+		name           string
+		opts           metav1.DeleteOptions
+		expectedErrors field.ErrorList
+	}{
+		{
+			name: "option is nil",
+			opts: metav1.DeleteOptions{
+				IgnoreStoreReadErrorWithClusterBreakingPotential: nil,
+				DryRun: []string{"All"},
+			},
+			expectedErrors: field.ErrorList{},
+		},
+		{
+			name: "option is false, PropagationPolicy is set",
+			opts: metav1.DeleteOptions{
+				IgnoreStoreReadErrorWithClusterBreakingPotential: ptr.To[bool](false),
+				DryRun:             []string{"All"},
+				PropagationPolicy:  ptr.To[metav1.DeletionPropagation](metav1.DeletePropagationBackground),
+				GracePeriodSeconds: ptr.To[int64](0),
+				Preconditions:      &metav1.Preconditions{},
+			},
+			expectedErrors: field.ErrorList{},
+		},
+		{
+			name: "option is false, OrphanDependents is set",
+			opts: metav1.DeleteOptions{
+				IgnoreStoreReadErrorWithClusterBreakingPotential: ptr.To[bool](false),
+				DryRun: []string{"All"},
+				//nolint:staticcheck // until it's being removed
+				OrphanDependents:   ptr.To[bool](true),
+				GracePeriodSeconds: ptr.To[int64](0),
+				Preconditions:      &metav1.Preconditions{},
+			},
+			expectedErrors: field.ErrorList{},
+		},
+		{
+			name: "option is true, PropagationPolicy is set",
+			opts: metav1.DeleteOptions{
+				IgnoreStoreReadErrorWithClusterBreakingPotential: ptr.To[bool](true),
+				DryRun:             []string{"All"},
+				PropagationPolicy:  ptr.To[metav1.DeletionPropagation](metav1.DeletePropagationBackground),
+				GracePeriodSeconds: ptr.To[int64](0),
+				Preconditions:      &metav1.Preconditions{},
+			},
+			expectedErrors: field.ErrorList{
+				field.Invalid(fieldPath, true, "cannot be set together with .dryRun"),
+				field.Invalid(fieldPath, true, "cannot be set together with .propagationPolicy"),
+				field.Invalid(fieldPath, true, "cannot be set together with .gracePeriodSeconds"),
+				field.Invalid(fieldPath, true, "cannot be set together with .preconditions"),
+			},
+		},
+		{
+			name: "option is true, OrphanDependents is set",
+			opts: metav1.DeleteOptions{
+				IgnoreStoreReadErrorWithClusterBreakingPotential: ptr.To[bool](true),
+				DryRun: []string{"All"},
+				//nolint:staticcheck // until it's being removed
+				OrphanDependents:   ptr.To[bool](true),
+				GracePeriodSeconds: ptr.To[int64](0),
+				Preconditions:      &metav1.Preconditions{},
+			},
+			expectedErrors: field.ErrorList{
+				field.Invalid(fieldPath, true, "cannot be set together with .dryRun"),
+				field.Invalid(fieldPath, true, "cannot be set together with .orphanDependents"),
+				field.Invalid(fieldPath, true, "cannot be set together with .gracePeriodSeconds"),
+				field.Invalid(fieldPath, true, "cannot be set together with .preconditions"),
+			},
+		},
+		{
+			name: "option is true, no other option is set",
+			opts: metav1.DeleteOptions{
+				IgnoreStoreReadErrorWithClusterBreakingPotential: ptr.To[bool](false),
+			},
+			expectedErrors: field.ErrorList{},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			errGot := ValidateDeleteOptions(&test.opts)
+			if !cmp.Equal(test.expectedErrors, errGot) {
+				t.Errorf("expected error(s) to match, diff: %s", cmp.Diff(test.expectedErrors, errGot))
+			}
+		})
+	}
+}
+
 func TestValidPatchOptions(t *testing.T) {
 	tests := []struct {
 		opts      metav1.PatchOptions

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/delete.go

@@ -32,6 +32,7 @@ import (
 	"k8s.io/apimachinery/pkg/apis/meta/v1/validation"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/validation/field"
 	"k8s.io/apiserver/pkg/admission"
 	"k8s.io/apiserver/pkg/audit"
 	"k8s.io/apiserver/pkg/endpoints/handlers/finisher"
@@ -44,6 +45,7 @@ import (
 	"k8s.io/apiserver/pkg/util/dryrun"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/component-base/tracing"
+	"k8s.io/utils/ptr"
 )
 
 // DeleteResource returns a function that will handle a resource deletion
@@ -265,6 +267,16 @@ func DeleteCollection(r rest.CollectionDeleter, checkBody bool, scope *RequestSc
 			scope.err(err, w, req)
 			return
 		}
+
+		if options != nil && ptr.Deref(options.IgnoreStoreReadErrorWithClusterBreakingPotential, true) {
+			fieldErrList := field.ErrorList{
+				field.Invalid(field.NewPath("ignoreStoreReadErrorWithClusterBreakingPotential"), true, "is not allowed with DELETECOLLECTION, try again after removing the option"),
+			}
+			err := errors.NewInvalid(schema.GroupKind{Group: metav1.GroupName, Kind: "DeleteOptions"}, "", fieldErrList)
+			scope.err(err, w, req)
+			return
+		}
+
 		options.TypeMeta.SetGroupVersionKind(metav1.SchemeGroupVersion.WithKind("DeleteOptions"))
 
 		admit = admission.WithAudit(admit)