Fix logic error in graceful deletion

pkg/api/rest/delete.go

@@ -81,7 +81,15 @@ func BeforeDelete(strategy RESTDeleteStrategy, ctx api.Context, obj runtime.Obje
 		// if we are already being deleted, we may only shorten the deletion grace period
 		// this means the object was gracefully deleted previously but deletionGracePeriodSeconds was not set,
 		// so we force deletion immediately
-		if objectMeta.DeletionGracePeriodSeconds == nil {
+		// IMPORTANT:
+		// The deletion operation happens in two phases.
+		// 1. Update to set DeletionGracePeriodSeconds and DeletionTimestamp
+		// 2. Delete the object from storage.
+		// If the update succeeds, but the delete fails (network error, internal storage error, etc.),
+		// a resource was previously left in a state that was non-recoverable.  We
+		// check if the existing stored resource has a grace period as 0 and if so
+		// attempt to delete immediately in order to recover from this scenario.
+		if objectMeta.DeletionGracePeriodSeconds == nil || *objectMeta.DeletionGracePeriodSeconds == 0 {
 			return false, false, nil
 		}
 		// only a shorter grace period may be provided by a user

pkg/registry/generic/registry/store_test.go

@@ -616,6 +616,43 @@ func TestStoreDelete(t *testing.T) {
 	}
 }
 
+// TestGracefulStoreCanDeleteIfExistingGracePeriodZero tests recovery from
+// race condition where the graceful delete is unable to complete
+// in prior operation, but the pod remains with deletion timestamp
+// and grace period set to 0.
+func TestGracefulStoreCanDeleteIfExistingGracePeriodZero(t *testing.T) {
+	deletionTimestamp := unversioned.NewTime(time.Now())
+	deletionGracePeriodSeconds := int64(0)
+	initialGeneration := int64(1)
+	pod := &api.Pod{
+		ObjectMeta: api.ObjectMeta{
+			Name:                       "foo",
+			Generation:                 initialGeneration,
+			DeletionGracePeriodSeconds: &deletionGracePeriodSeconds,
+			DeletionTimestamp:          &deletionTimestamp,
+		},
+		Spec: api.PodSpec{NodeName: "machine"},
+	}
+
+	testContext := api.WithNamespace(api.NewContext(), "test")
+	destroyFunc, registry := NewTestGenericStoreRegistry(t)
+	registry.EnableGarbageCollection = false
+	defaultDeleteStrategy := testRESTStrategy{api.Scheme, api.SimpleNameGenerator, true, false, true}
+	registry.DeleteStrategy = testGracefulStrategy{defaultDeleteStrategy}
+	defer destroyFunc()
+
+	graceful, gracefulPending, err := rest.BeforeDelete(registry.DeleteStrategy, testContext, pod, api.NewDeleteOptions(0))
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if graceful {
+		t.Fatalf("graceful should be false if object has DeletionTimestamp and DeletionGracePeriodSeconds is 0")
+	}
+	if gracefulPending {
+		t.Fatalf("gracefulPending should be false if object has DeletionTimestamp and DeletionGracePeriodSeconds is 0")
+	}
+}
+
 func TestGracefulStoreHandleFinalizers(t *testing.T) {
 	initialGeneration := int64(1)
 	podWithFinalizer := &api.Pod{