Enable finalizers independent of GC enablement

Decouple finalizer processing from garbage collection configuration.
Finalizers should be effective even when garbage collection is disabled
for a given store.

Fixes https://github.com/kubernetes/kubernetes/issues/50528.

staging/src/k8s.io/apiserver/pkg/registry/generic/registry/store.go

@@ -442,12 +442,8 @@ func (e *Store) WaitForInitialized(ctx genericapirequest.Context, obj runtime.Ob
 
 // shouldDeleteDuringUpdate checks if a Update is removing all the object's
 // finalizers. If so, it further checks if the object's
-// DeletionGracePeriodSeconds is 0. If so, it returns true. If garbage collection
+// DeletionGracePeriodSeconds is 0.
-// is disabled it always returns false.
 func (e *Store) shouldDeleteDuringUpdate(ctx genericapirequest.Context, key string, obj, existing runtime.Object) bool {
-	if !e.EnableGarbageCollection {
-		return false
-	}
 	newMeta, err := meta.Accessor(obj)
 	if err != nil {
 		utilruntime.HandleError(err)
@@ -765,9 +761,16 @@ func shouldDeleteDependents(e *Store, accessor metav1.Object, options *metav1.De
 	return false
 }
 
-// deletionFinalizers returns the deletion finalizers we should set on the object and a bool indicate they did or
+// deletionFinalizers returns the deletion finalizers we should set on the
-// did not change.
+// object and a bool indicate they did or did not change.
+//
+// Because the finalizers created here are only cleared by the garbage
+// collector, deletionFinalizers always returns false when garbage collection is
+// disabled for the Store.
 func deletionFinalizers(e *Store, accessor metav1.Object, options *metav1.DeleteOptions) (bool, []string) {
+	if !e.EnableGarbageCollection {
+		return false, []string{}
+	}
 	shouldOrphan := shouldOrphanDependents(e, accessor, options)
 	shouldDeleteDependentInForeground := shouldDeleteDependents(e, accessor, options)
 	newFinalizers := []string{}
@@ -816,72 +819,6 @@ func markAsDeleting(obj runtime.Object) (err error) {
 	return nil
 }
 
-// updateForGracefulDeletion and updateForGracefulDeletionAndFinalizers both
-// implement deletion flows for graceful deletion.  Graceful deletion is
-// implemented as setting the deletion timestamp in an update.  If the
-// implementation of graceful deletion is changed, both of these methods
-// should be changed together.
-
-// updateForGracefulDeletion updates the given object for graceful deletion by
-// setting the deletion timestamp and grace period seconds and returns:
-//
-// 1. an error
-// 2. a boolean indicating that the object was not found, but it should be
-//    ignored
-// 3. a boolean indicating that the object's grace period is exhausted and it
-//    should be deleted immediately
-// 4. a new output object with the state that was updated
-// 5. a copy of the last existing state of the object
-func (e *Store) updateForGracefulDeletion(ctx genericapirequest.Context, name, key string, options *metav1.DeleteOptions, preconditions storage.Preconditions, in runtime.Object) (err error, ignoreNotFound, deleteImmediately bool, out, lastExisting runtime.Object) {
-	lastGraceful := int64(0)
-	out = e.NewFunc()
-	err = e.Storage.GuaranteedUpdate(
-		ctx,
-		key,
-		out,
-		false, /* ignoreNotFound */
-		&preconditions,
-		storage.SimpleUpdate(func(existing runtime.Object) (runtime.Object, error) {
-			graceful, pendingGraceful, err := rest.BeforeDelete(e.DeleteStrategy, ctx, existing, options)
-			if err != nil {
-				return nil, err
-			}
-			if pendingGraceful {
-				return nil, errAlreadyDeleting
-			}
-			if !graceful {
-				return nil, errDeleteNow
-			}
-			lastGraceful = *options.GracePeriodSeconds
-			lastExisting = existing
-			return existing, nil
-		}),
-	)
-	switch err {
-	case nil:
-		if lastGraceful > 0 {
-			return nil, false, false, out, lastExisting
-		}
-		// If we are here, the registry supports grace period mechanism and
-		// we are intentionally delete gracelessly. In this case, we may
-		// enter a race with other k8s components. If other component wins
-		// the race, the object will not be found, and we should tolerate
-		// the NotFound error. See
-		// https://github.com/kubernetes/kubernetes/issues/19403 for
-		// details.
-		return nil, true, true, out, lastExisting
-	case errDeleteNow:
-		// we've updated the object to have a zero grace period, or it's already at 0, so
-		// we should fall through and truly delete the object.
-		return nil, false, true, out, lastExisting
-	case errAlreadyDeleting:
-		out, err = e.finalizeDelete(ctx, in, true)
-		return err, false, false, out, lastExisting
-	default:
-		return storeerr.InterpretUpdateError(err, e.qualifiedResourceFromContext(ctx), name), false, false, out, lastExisting
-	}
-}
-
 // updateForGracefulDeletionAndFinalizers updates the given object for
 // graceful deletion and finalization by setting the deletion timestamp and
 // grace period seconds (graceful deletion) and updating the list of
@@ -1013,18 +950,12 @@ func (e *Store) Delete(ctx genericapirequest.Context, name string, options *meta
 
 	// Handle combinations of graceful deletion and finalization by issuing
 	// the correct updates.
-	if e.EnableGarbageCollection {
 	shouldUpdateFinalizers, _ := deletionFinalizers(e, accessor, options)
 	// TODO: remove the check, because we support no-op updates now.
 	if graceful || pendingFinalizers || shouldUpdateFinalizers {
 		err, ignoreNotFound, deleteImmediately, out, lastExisting = e.updateForGracefulDeletionAndFinalizers(ctx, name, key, options, preconditions, obj)
 	}
-	} else {
+
-		// TODO: remove the check on graceful, because we support no-op updates now.
-		if graceful {
-			err, ignoreNotFound, deleteImmediately, out, lastExisting = e.updateForGracefulDeletion(ctx, name, key, options, preconditions, obj)
-		}
-	}
 	// !deleteImmediately covers all cases where err != nil. We keep both to be future-proof.
 	if !deleteImmediately || err != nil {
 		return out, false, err

staging/src/k8s.io/apiserver/pkg/registry/generic/registry/store_test.go

@@ -931,10 +931,15 @@ func TestGracefulStoreHandleFinalizers(t *testing.T) {
 
 	testContext := genericapirequest.WithNamespace(genericapirequest.NewContext(), "test")
 	destroyFunc, registry := NewTestGenericStoreRegistry(t)
-	registry.EnableGarbageCollection = true
 	defaultDeleteStrategy := testRESTStrategy{scheme, names.SimpleNameGenerator, true, false, true}
 	registry.DeleteStrategy = testGracefulStrategy{defaultDeleteStrategy}
 	defer destroyFunc()
+
+	gcStates := []bool{true, false}
+	for _, gcEnabled := range gcStates {
+		t.Logf("garbage collection enabled: %t", gcEnabled)
+		registry.EnableGarbageCollection = gcEnabled
+
 		// create pod
 		_, err := registry.Create(testContext, podWithFinalizer, false)
 		if err != nil {
@@ -983,6 +988,7 @@ func TestGracefulStoreHandleFinalizers(t *testing.T) {
 			t.Fatalf("Unexpected error: %v", err)
 		}
 	}
+}
 
 func TestFailedInitializationStoreUpdate(t *testing.T) {
 	initialGeneration := int64(1)
@@ -1030,8 +1036,13 @@ func TestNonGracefulStoreHandleFinalizers(t *testing.T) {
 
 	testContext := genericapirequest.WithNamespace(genericapirequest.NewContext(), "test")
 	destroyFunc, registry := NewTestGenericStoreRegistry(t)
-	registry.EnableGarbageCollection = true
+
 	defer destroyFunc()
+	gcStates := []bool{true, false}
+	for _, gcEnabled := range gcStates {
+		t.Logf("garbage collection enabled: %t", gcEnabled)
+		registry.EnableGarbageCollection = gcEnabled
+
 		// create pod
 		_, err := registry.Create(testContext, podWithFinalizer, false)
 		if err != nil {
@@ -1099,6 +1110,7 @@ func TestNonGracefulStoreHandleFinalizers(t *testing.T) {
 			t.Errorf("Unexpected error: %v", err)
 		}
 	}
+}
 
 func TestStoreDeleteWithOrphanDependents(t *testing.T) {
 	initialGeneration := int64(1)