Merge pull request #58743 from NickrenREN/pv-protection

Automatic merge from submit-queue. If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>.

Postpone PV deletion with finalizer when it is being used

Postpone PV deletion if it is bound to a PVC

xref: https://github.com/kubernetes/community/pull/1608


**Which issue(s) this PR fixes** *(optional, in `fixes #<issue number>(, fixes #<issue_number>, ...)` format, will close the issue(s) when PR gets merged)*:
Fixes #33355

**Special notes for your reviewer**:

**Release note**:
```release-note
Postpone PV deletion when it is being bound to a PVC
```

WIP, assign to myself first

/assign @NickrenREN

cluster/centos/config-default.sh

@@ -120,7 +120,7 @@ export FLANNEL_NET=${FLANNEL_NET:-"172.16.0.0/16"}
 
 # Admission Controllers to invoke prior to persisting objects in cluster
 # If we included ResourceQuota, we should keep it at the end of the list to prevent incrementing quota usage prematurely.
-export ADMISSION_CONTROL=${ADMISSION_CONTROL:-"Initializers,NamespaceLifecycle,LimitRanger,ServiceAccount,PersistentVolumeClaimResize,DefaultTolerationSeconds,Priority,PVCProtection,ResourceQuota"}
+export ADMISSION_CONTROL=${ADMISSION_CONTROL:-"Initializers,NamespaceLifecycle,LimitRanger,ServiceAccount,PersistentVolumeClaimResize,DefaultTolerationSeconds,Priority,StorageProtection,ResourceQuota"}
 
 # Extra options to set on the Docker command line.
 # This is useful for setting --insecure-registry for local registries.

cluster/gce/config-default.sh

@@ -298,7 +298,7 @@ if [[ -n "${GCE_GLBC_IMAGE:-}" ]]; then
 fi
 
 # Admission Controllers to invoke prior to persisting objects in cluster
-ADMISSION_CONTROL=Initializers,NamespaceLifecycle,LimitRanger,ServiceAccount,PersistentVolumeLabel,DefaultStorageClass,PersistentVolumeClaimResize,DefaultTolerationSeconds,NodeRestriction,Priority,PVCProtection
+ADMISSION_CONTROL=Initializers,NamespaceLifecycle,LimitRanger,ServiceAccount,PersistentVolumeLabel,DefaultStorageClass,PersistentVolumeClaimResize,DefaultTolerationSeconds,NodeRestriction,Priority,StorageProtection
 
 if [[ "${ENABLE_POD_SECURITY_POLICY:-}" == "true" ]]; then
   ADMISSION_CONTROL="${ADMISSION_CONTROL},PodSecurityPolicy"

cmd/kube-controller-manager/app/BUILD

@@ -76,6 +76,7 @@ go_library(
         "//pkg/controller/volume/expand:go_default_library",
         "//pkg/controller/volume/persistentvolume:go_default_library",
         "//pkg/controller/volume/pvcprotection:go_default_library",
+        "//pkg/controller/volume/pvprotection:go_default_library",
         "//pkg/features:go_default_library",
         "//pkg/quota/generic:go_default_library",
         "//pkg/quota/install:go_default_library",

cmd/kube-controller-manager/app/controllermanager.go

@@ -384,6 +384,7 @@ func NewControllerInitializers(loopMode ControllerLoopMode) map[string]InitFunc
 	controllers["persistentvolume-expander"] = startVolumeExpandController
 	controllers["clusterrole-aggregation"] = startClusterRoleAggregrationController
 	controllers["pvc-protection"] = startPVCProtectionController
+	controllers["pv-protection"] = startPVProtectionController
 
 	return controllers
 }

cmd/kube-controller-manager/app/core.go

@@ -55,6 +55,7 @@ import (
 	"k8s.io/kubernetes/pkg/controller/volume/expand"
 	persistentvolumecontroller "k8s.io/kubernetes/pkg/controller/volume/persistentvolume"
 	"k8s.io/kubernetes/pkg/controller/volume/pvcprotection"
+	"k8s.io/kubernetes/pkg/controller/volume/pvprotection"
 	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/quota/generic"
 	quotainstall "k8s.io/kubernetes/pkg/quota/install"
@@ -393,7 +394,7 @@ func startGarbageCollectorController(ctx ControllerContext) (bool, error) {
 }
 
 func startPVCProtectionController(ctx ControllerContext) (bool, error) {
-	if utilfeature.DefaultFeatureGate.Enabled(features.PVCProtection) {
+	if utilfeature.DefaultFeatureGate.Enabled(features.StorageProtection) {
 		go pvcprotection.NewPVCProtectionController(
 			ctx.InformerFactory.Core().V1().PersistentVolumeClaims(),
 			ctx.InformerFactory.Core().V1().Pods(),
@@ -403,3 +404,14 @@ func startPVCProtectionController(ctx ControllerContext) (bool, error) {
 	}
 	return false, nil
 }
+
+func startPVProtectionController(ctx ControllerContext) (bool, error) {
+	if utilfeature.DefaultFeatureGate.Enabled(features.StorageProtection) {
+		go pvprotection.NewPVProtectionController(
+			ctx.InformerFactory.Core().V1().PersistentVolumes(),
+			ctx.ClientBuilder.ClientOrDie("pv-protection-controller"),
+		).Run(1, ctx.Stop)
+		return true, nil
+	}
+	return false, nil
+}

pkg/controller/BUILD

@@ -136,6 +136,7 @@ filegroup(
         "//pkg/controller/volume/expand:all-srcs",
         "//pkg/controller/volume/persistentvolume:all-srcs",
         "//pkg/controller/volume/pvcprotection:all-srcs",
+        "//pkg/controller/volume/pvprotection:all-srcs",
     ],
     tags = ["automanaged"],
 )

pkg/controller/volume/persistentvolume/index.go

@@ -169,6 +169,13 @@ func findMatchingVolume(
 			continue
 		}
 
+		// check if PV's DeletionTimeStamp is set, if so, skip this volume.
+		if utilfeature.DefaultFeatureGate.Enabled(features.StorageProtection) {
+			if volume.ObjectMeta.DeletionTimestamp != nil {
+				continue
+			}
+		}
+
 		nodeAffinityValid := true
 		if node != nil {
 			// Scheduler path, check that the PV NodeAffinity

pkg/controller/volume/persistentvolume/index_test.go

@@ -854,18 +854,22 @@ func createTestVolOrderedIndex(pv *v1.PersistentVolume) persistentVolumeOrderedI
 	return volFile
 }
 
-func toggleBlockVolumeFeature(toggleFlag bool, t *testing.T) {
+func toggleFeature(toggleFlag bool, featureName string, t *testing.T) {
+	var valueStr string
 	if toggleFlag {
-		// Enable alpha feature BlockVolume
+		// Enable feature
-		err := utilfeature.DefaultFeatureGate.Set("BlockVolume=true")
+		valueStr = featureName + "=true"
+		err := utilfeature.DefaultFeatureGate.Set(valueStr)
 		if err != nil {
-			t.Errorf("Failed to enable feature gate for BlockVolume: %v", err)
+			t.Errorf("Failed to enable feature gate for %s: %v", featureName, err)
 			return
 		}
 	} else {
-		err := utilfeature.DefaultFeatureGate.Set("BlockVolume=false")
+		// Disable feature
+		valueStr = featureName + "=false"
+		err := utilfeature.DefaultFeatureGate.Set(valueStr)
 		if err != nil {
-			t.Errorf("Failed to disable feature gate for BlockVolume: %v", err)
+			t.Errorf("Failed to disable feature gate for %s: %v", featureName, err)
 			return
 		}
 	}
@@ -935,7 +939,7 @@ func TestAlphaVolumeModeCheck(t *testing.T) {
 	}
 
 	for name, scenario := range scenarios {
-		toggleBlockVolumeFeature(scenario.enableBlock, t)
+		toggleFeature(scenario.enableBlock, "BlockVolume", t)
 		expectedMisMatch, err := checkVolumeModeMisMatches(&scenario.pvc.Spec, &scenario.vol.Spec)
 		if err != nil {
 			t.Errorf("Unexpected failure for checkVolumeModeMisMatches: %v", err)
@@ -950,7 +954,7 @@ func TestAlphaVolumeModeCheck(t *testing.T) {
 	}
 
 	// make sure feature gate is turned off
-	toggleBlockVolumeFeature(false, t)
+	toggleFeature(false, "BlockVolume", t)
 }
 
 func TestAlphaFilteringVolumeModes(t *testing.T) {
@@ -1028,7 +1032,7 @@ func TestAlphaFilteringVolumeModes(t *testing.T) {
 	}
 
 	for name, scenario := range scenarios {
-		toggleBlockVolumeFeature(scenario.enableBlock, t)
+		toggleFeature(scenario.enableBlock, "BlockVolume", t)
 		pvmatch, err := scenario.vol.findBestMatchForClaim(scenario.pvc, false)
 		// expected to match but either got an error or no returned pvmatch
 		if pvmatch == nil && scenario.isExpectedMatch {
@@ -1047,7 +1051,135 @@ func TestAlphaFilteringVolumeModes(t *testing.T) {
 	}
 
 	// make sure feature gate is turned off
-	toggleBlockVolumeFeature(false, t)
+	toggleFeature(false, "BlockVolume", t)
+}
+
+func TestAlphaStorageProtectionFiltering(t *testing.T) {
+	pv := &v1.PersistentVolume{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        "pv1",
+			Annotations: map[string]string{},
+		},
+		Spec: v1.PersistentVolumeSpec{
+			Capacity:               v1.ResourceList{v1.ResourceName(v1.ResourceStorage): resource.MustParse("1G")},
+			PersistentVolumeSource: v1.PersistentVolumeSource{HostPath: &v1.HostPathVolumeSource{}},
+			AccessModes:            []v1.PersistentVolumeAccessMode{v1.ReadWriteOnce},
+		},
+	}
+
+	pvToDelete := pv.DeepCopy()
+	now := metav1.Now()
+	pvToDelete.ObjectMeta.DeletionTimestamp = &now
+
+	pvc := &v1.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "pvc1",
+			Namespace: "myns",
+		},
+		Spec: v1.PersistentVolumeClaimSpec{
+			AccessModes: []v1.PersistentVolumeAccessMode{v1.ReadWriteOnce},
+			Resources:   v1.ResourceRequirements{Requests: v1.ResourceList{v1.ResourceName(v1.ResourceStorage): resource.MustParse("1G")}},
+		},
+	}
+
+	satisfyingTestCases := map[string]struct {
+		isExpectedMatch         bool
+		vol                     *v1.PersistentVolume
+		pvc                     *v1.PersistentVolumeClaim
+		enableStorageProtection bool
+	}{
+		"feature enabled - pv deletionTimeStamp not set": {
+			isExpectedMatch: true,
+			vol:             pv,
+			pvc:             pvc,
+			enableStorageProtection: true,
+		},
+		"feature enabled - pv deletionTimeStamp set": {
+			isExpectedMatch: false,
+			vol:             pvToDelete,
+			pvc:             pvc,
+			enableStorageProtection: true,
+		},
+		"feature disabled - pv deletionTimeStamp not set": {
+			isExpectedMatch: true,
+			vol:             pv,
+			pvc:             pvc,
+			enableStorageProtection: false,
+		},
+		"feature disabled - pv deletionTimeStamp set": {
+			isExpectedMatch: true,
+			vol:             pvToDelete,
+			pvc:             pvc,
+			enableStorageProtection: false,
+		},
+	}
+
+	for name, testCase := range satisfyingTestCases {
+		toggleFeature(testCase.enableStorageProtection, "StorageProtection", t)
+		err := checkVolumeSatisfyClaim(testCase.vol, testCase.pvc)
+		// expected to match but got an error
+		if err != nil && testCase.isExpectedMatch {
+			t.Errorf("%s: expected to match but got an error: %v", name, err)
+		}
+		// not expected to match but did
+		if err == nil && !testCase.isExpectedMatch {
+			t.Errorf("%s: not expected to match but did", name)
+		}
+
+	}
+
+	filteringTestCases := map[string]struct {
+		isExpectedMatch         bool
+		vol                     persistentVolumeOrderedIndex
+		pvc                     *v1.PersistentVolumeClaim
+		enableStorageProtection bool
+	}{
+		"feature enabled - pv deletionTimeStamp not set": {
+			isExpectedMatch: true,
+			vol:             createTestVolOrderedIndex(pv),
+			pvc:             pvc,
+			enableStorageProtection: true,
+		},
+		"feature enabled - pv deletionTimeStamp set": {
+			isExpectedMatch: false,
+			vol:             createTestVolOrderedIndex(pvToDelete),
+			pvc:             pvc,
+			enableStorageProtection: true,
+		},
+		"feature disabled - pv deletionTimeStamp not set": {
+			isExpectedMatch: true,
+			vol:             createTestVolOrderedIndex(pv),
+			pvc:             pvc,
+			enableStorageProtection: false,
+		},
+		"feature disabled - pv deletionTimeStamp set": {
+			isExpectedMatch: true,
+			vol:             createTestVolOrderedIndex(pvToDelete),
+			pvc:             pvc,
+			enableStorageProtection: false,
+		},
+	}
+	for name, testCase := range filteringTestCases {
+		toggleFeature(testCase.enableStorageProtection, "StorageProtection", t)
+		pvmatch, err := testCase.vol.findBestMatchForClaim(testCase.pvc, false)
+		// expected to match but either got an error or no returned pvmatch
+		if pvmatch == nil && testCase.isExpectedMatch {
+			t.Errorf("Unexpected failure for testcase, no matching volume: %s", name)
+		}
+		if err != nil && testCase.isExpectedMatch {
+			t.Errorf("Unexpected failure for testcase: %s - %+v", name, err)
+		}
+		// expected to not match but either got an error or a returned pvmatch
+		if pvmatch != nil && !testCase.isExpectedMatch {
+			t.Errorf("Unexpected failure for testcase, expected no matching volume: %s", name)
+		}
+		if err != nil && !testCase.isExpectedMatch {
+			t.Errorf("Unexpected failure for testcase: %s - %+v", name, err)
+		}
+	}
+
+	// make sure feature gate is turned off
+	toggleFeature(false, "StorageProtection", t)
 }
 
 func TestFindingPreboundVolumes(t *testing.T) {

pkg/controller/volume/persistentvolume/pv_controller.go

@@ -233,6 +233,14 @@ func (ctrl *PersistentVolumeController) syncClaim(claim *v1.PersistentVolumeClai
 func checkVolumeSatisfyClaim(volume *v1.PersistentVolume, claim *v1.PersistentVolumeClaim) error {
 	requestedQty := claim.Spec.Resources.Requests[v1.ResourceName(v1.ResourceStorage)]
 	requestedSize := requestedQty.Value()
+
+	// check if PV's DeletionTimeStamp is set, if so, return error.
+	if utilfeature.DefaultFeatureGate.Enabled(features.StorageProtection) {
+		if volume.ObjectMeta.DeletionTimestamp != nil {
+			return fmt.Errorf("the volume is marked for deletion")
+		}
+	}
+
 	volumeQty := volume.Spec.Capacity[v1.ResourceStorage]
 	volumeSize := volumeQty.Value()
 	if volumeSize < requestedSize {

pkg/controller/volume/pvprotection/BUILD

@@ -0,0 +1,60 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["pv_protection_controller.go"],
+    importpath = "k8s.io/kubernetes/pkg/controller/volume/pvprotection",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/controller:go_default_library",
+        "//pkg/util/metrics:go_default_library",
+        "//pkg/util/slice:go_default_library",
+        "//pkg/volume/util:go_default_library",
+        "//vendor/github.com/golang/glog:go_default_library",
+        "//vendor/k8s.io/api/core/v1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/util/runtime:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
+        "//vendor/k8s.io/client-go/informers/core/v1:go_default_library",
+        "//vendor/k8s.io/client-go/kubernetes:go_default_library",
+        "//vendor/k8s.io/client-go/listers/core/v1:go_default_library",
+        "//vendor/k8s.io/client-go/tools/cache:go_default_library",
+        "//vendor/k8s.io/client-go/util/workqueue:go_default_library",
+    ],
+)
+
+filegroup(
+    name = "package-srcs",
+    srcs = glob(["**"]),
+    tags = ["automanaged"],
+    visibility = ["//visibility:private"],
+)
+
+filegroup(
+    name = "all-srcs",
+    srcs = [":package-srcs"],
+    tags = ["automanaged"],
+    visibility = ["//visibility:public"],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["pv_protection_controller_test.go"],
+    embed = [":go_default_library"],
+    importpath = "k8s.io/kubernetes/pkg/controller/volume/pvprotection",
+    deps = [
+        "//pkg/controller:go_default_library",
+        "//pkg/volume/util:go_default_library",
+        "//vendor/github.com/davecgh/go-spew/spew:go_default_library",
+        "//vendor/github.com/golang/glog:go_default_library",
+        "//vendor/k8s.io/api/core/v1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/api/meta:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
+        "//vendor/k8s.io/client-go/informers:go_default_library",
+        "//vendor/k8s.io/client-go/kubernetes/fake:go_default_library",
+        "//vendor/k8s.io/client-go/testing:go_default_library",
+    ],
+)

pkg/controller/volume/pvprotection/pv_protection_controller.go

@@ -0,0 +1,208 @@
+/*
+Copyright 2018 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 pvprotection
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/golang/glog"
+	"k8s.io/api/core/v1"
+	apierrs "k8s.io/apimachinery/pkg/api/errors"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
+	coreinformers "k8s.io/client-go/informers/core/v1"
+	clientset "k8s.io/client-go/kubernetes"
+	corelisters "k8s.io/client-go/listers/core/v1"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/workqueue"
+	"k8s.io/kubernetes/pkg/controller"
+	"k8s.io/kubernetes/pkg/util/metrics"
+	"k8s.io/kubernetes/pkg/util/slice"
+	volumeutil "k8s.io/kubernetes/pkg/volume/util"
+)
+
+// Controller is controller that removes PVProtectionFinalizer
+// from PVs that are not bound to PVCs.
+type Controller struct {
+	client clientset.Interface
+
+	pvLister       corelisters.PersistentVolumeLister
+	pvListerSynced cache.InformerSynced
+
+	queue workqueue.RateLimitingInterface
+}
+
+// NewPVProtectionController returns a new *Controller.
+func NewPVProtectionController(pvInformer coreinformers.PersistentVolumeInformer, cl clientset.Interface) *Controller {
+	e := &Controller{
+		client: cl,
+		queue:  workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "pvprotection"),
+	}
+	if cl != nil && cl.CoreV1().RESTClient().GetRateLimiter() != nil {
+		metrics.RegisterMetricAndTrackRateLimiterUsage("persistentvolume_protection_controller", cl.CoreV1().RESTClient().GetRateLimiter())
+	}
+
+	e.pvLister = pvInformer.Lister()
+	e.pvListerSynced = pvInformer.Informer().HasSynced
+	pvInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: e.pvAddedUpdated,
+		UpdateFunc: func(old, new interface{}) {
+			e.pvAddedUpdated(new)
+		},
+	})
+
+	return e
+}
+
+// Run runs the controller goroutines.
+func (c *Controller) Run(workers int, stopCh <-chan struct{}) {
+	defer utilruntime.HandleCrash()
+	defer c.queue.ShutDown()
+
+	glog.Infof("Starting PV protection controller")
+	defer glog.Infof("Shutting down PV protection controller")
+
+	if !controller.WaitForCacheSync("PV protection", stopCh, c.pvListerSynced) {
+		return
+	}
+
+	for i := 0; i < workers; i++ {
+		go wait.Until(c.runWorker, time.Second, stopCh)
+	}
+
+	<-stopCh
+}
+
+func (c *Controller) runWorker() {
+	for c.processNextWorkItem() {
+	}
+}
+
+// processNextWorkItem deals with one pvcKey off the queue.  It returns false when it's time to quit.
+func (c *Controller) processNextWorkItem() bool {
+	pvKey, quit := c.queue.Get()
+	if quit {
+		return false
+	}
+	defer c.queue.Done(pvKey)
+
+	pvName := pvKey.(string)
+
+	err := c.processPV(pvName)
+	if err == nil {
+		c.queue.Forget(pvKey)
+		return true
+	}
+
+	utilruntime.HandleError(fmt.Errorf("PV %v failed with : %v", pvKey, err))
+	c.queue.AddRateLimited(pvKey)
+
+	return true
+}
+
+func (c *Controller) processPV(pvName string) error {
+	glog.V(4).Infof("Processing PV %s", pvName)
+	startTime := time.Now()
+	defer func() {
+		glog.V(4).Infof("Finished processing PV %s (%v)", pvName, time.Now().Sub(startTime))
+	}()
+
+	pv, err := c.pvLister.Get(pvName)
+	if apierrs.IsNotFound(err) {
+		glog.V(4).Infof("PV %s not found, ignoring", pvName)
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+
+	if isDeletionCandidate(pv) {
+		// PV should be deleted. Check if it's used and remove finalizer if
+		// it's not.
+		isUsed := c.isBeingUsed(pv)
+		if !isUsed {
+			return c.removeFinalizer(pv)
+		}
+	}
+
+	if needToAddFinalizer(pv) {
+		// PV is not being deleted -> it should have the finalizer. The
+		// finalizer should be added by admission plugin, this is just to add
+		// the finalizer to old PVs that were created before the admission
+		// plugin was enabled.
+		return c.addFinalizer(pv)
+	}
+	return nil
+}
+
+func (c *Controller) addFinalizer(pv *v1.PersistentVolume) error {
+	pvClone := pv.DeepCopy()
+	pvClone.ObjectMeta.Finalizers = append(pvClone.ObjectMeta.Finalizers, volumeutil.PVProtectionFinalizer)
+	_, err := c.client.CoreV1().PersistentVolumes().Update(pvClone)
+	if err != nil {
+		glog.V(3).Infof("Error adding protection finalizer to PV %s: %v", pv.Name)
+		return err
+	}
+	glog.V(3).Infof("Added protection finalizer to PV %s", pv.Name)
+	return nil
+}
+
+func (c *Controller) removeFinalizer(pv *v1.PersistentVolume) error {
+	pvClone := pv.DeepCopy()
+	pvClone.ObjectMeta.Finalizers = slice.RemoveString(pvClone.ObjectMeta.Finalizers, volumeutil.PVProtectionFinalizer, nil)
+	_, err := c.client.CoreV1().PersistentVolumes().Update(pvClone)
+	if err != nil {
+		glog.V(3).Infof("Error removing protection finalizer from PV %s: %v", pv.Name, err)
+		return err
+	}
+	glog.V(3).Infof("Removed protection finalizer from PV %s", pv.Name)
+	return nil
+}
+
+func (c *Controller) isBeingUsed(pv *v1.PersistentVolume) bool {
+	// check if PV is being bound to a PVC by its status
+	// the status will be updated by PV controller
+	if pv.Status.Phase == v1.VolumeBound {
+		// the PV is being used now
+		return true
+	}
+
+	return false
+}
+
+// pvAddedUpdated reacts to pv added/updated events
+func (c *Controller) pvAddedUpdated(obj interface{}) {
+	pv, ok := obj.(*v1.PersistentVolume)
+	if !ok {
+		utilruntime.HandleError(fmt.Errorf("PV informer returned non-PV object: %#v", obj))
+		return
+	}
+	glog.V(4).Infof("Got event on PV %s", pv.Name)
+
+	if needToAddFinalizer(pv) || isDeletionCandidate(pv) {
+		c.queue.Add(pv.Name)
+	}
+}
+
+func isDeletionCandidate(pv *v1.PersistentVolume) bool {
+	return pv.ObjectMeta.DeletionTimestamp != nil && slice.ContainsString(pv.ObjectMeta.Finalizers, volumeutil.PVProtectionFinalizer, nil)
+}
+
+func needToAddFinalizer(pv *v1.PersistentVolume) bool {
+	return pv.ObjectMeta.DeletionTimestamp == nil && !slice.ContainsString(pv.ObjectMeta.Finalizers, volumeutil.PVProtectionFinalizer, nil)
+}

pkg/controller/volume/pvprotection/pv_protection_controller_test.go

@@ -0,0 +1,257 @@
+/*
+Copyright 2018 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 pvprotection
+
+import (
+	"errors"
+	"reflect"
+	"testing"
+	"time"
+
+	"github.com/davecgh/go-spew/spew"
+	"github.com/golang/glog"
+
+	"k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes/fake"
+	clienttesting "k8s.io/client-go/testing"
+	"k8s.io/kubernetes/pkg/controller"
+	volumeutil "k8s.io/kubernetes/pkg/volume/util"
+)
+
+const defaultPVName = "default-pv"
+
+type reaction struct {
+	verb      string
+	resource  string
+	reactorfn clienttesting.ReactionFunc
+}
+
+func pv() *v1.PersistentVolume {
+	return &v1.PersistentVolume{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: defaultPVName,
+		},
+	}
+}
+
+func boundPV() *v1.PersistentVolume {
+	return &v1.PersistentVolume{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: defaultPVName,
+		},
+		Status: v1.PersistentVolumeStatus{
+			Phase: v1.VolumeBound,
+		},
+	}
+}
+
+func withProtectionFinalizer(pv *v1.PersistentVolume) *v1.PersistentVolume {
+	pv.Finalizers = append(pv.Finalizers, volumeutil.PVProtectionFinalizer)
+	return pv
+}
+
+func generateUpdateErrorFunc(t *testing.T, failures int) clienttesting.ReactionFunc {
+	i := 0
+	return func(action clienttesting.Action) (bool, runtime.Object, error) {
+		i++
+		if i <= failures {
+			// Update fails
+			update, ok := action.(clienttesting.UpdateAction)
+
+			if !ok {
+				t.Fatalf("Reactor got non-update action: %+v", action)
+			}
+			acc, _ := meta.Accessor(update.GetObject())
+			return true, nil, apierrors.NewForbidden(update.GetResource().GroupResource(), acc.GetName(), errors.New("Mock error"))
+		}
+		// Update succeeds
+		return false, nil, nil
+	}
+}
+
+func deleted(pv *v1.PersistentVolume) *v1.PersistentVolume {
+	pv.DeletionTimestamp = &metav1.Time{}
+	return pv
+}
+
+func TestPVProtectionController(t *testing.T) {
+	pvVer := schema.GroupVersionResource{
+		Group:    v1.GroupName,
+		Version:  "v1",
+		Resource: "persistentvolumes",
+	}
+	tests := []struct {
+		name string
+		// Object to insert into fake kubeclient before the test starts.
+		initialObjects []runtime.Object
+		// Optional client reactors.
+		reactors []reaction
+		// PV event to simulate. This PV will be automatically added to
+		// initalObjects.
+		updatedPV *v1.PersistentVolume
+		// List of expected kubeclient actions that should happen during the
+		// test.
+		expectedActions []clienttesting.Action
+	}{
+		// PV events
+		//
+		{
+			name:      "PV without finalizer -> finalizer is added",
+			updatedPV: pv(),
+			expectedActions: []clienttesting.Action{
+				clienttesting.NewUpdateAction(pvVer, "", withProtectionFinalizer(pv())),
+			},
+		},
+		{
+			name:            "PVC with finalizer -> no action",
+			updatedPV:       withProtectionFinalizer(pv()),
+			expectedActions: []clienttesting.Action{},
+		},
+		{
+			name:      "saving PVC finalizer fails -> controller retries",
+			updatedPV: pv(),
+			reactors: []reaction{
+				{
+					verb:      "update",
+					resource:  "persistentvolumes",
+					reactorfn: generateUpdateErrorFunc(t, 2 /* update fails twice*/),
+				},
+			},
+			expectedActions: []clienttesting.Action{
+				// This fails
+				clienttesting.NewUpdateAction(pvVer, "", withProtectionFinalizer(pv())),
+				// This fails too
+				clienttesting.NewUpdateAction(pvVer, "", withProtectionFinalizer(pv())),
+				// This succeeds
+				clienttesting.NewUpdateAction(pvVer, "", withProtectionFinalizer(pv())),
+			},
+		},
+		{
+			name:      "deleted PV with finalizer -> finalizer is removed",
+			updatedPV: deleted(withProtectionFinalizer(pv())),
+			expectedActions: []clienttesting.Action{
+				clienttesting.NewUpdateAction(pvVer, "", deleted(pv())),
+			},
+		},
+		{
+			name:      "finalizer removal fails -> controller retries",
+			updatedPV: deleted(withProtectionFinalizer(pv())),
+			reactors: []reaction{
+				{
+					verb:      "update",
+					resource:  "persistentvolumes",
+					reactorfn: generateUpdateErrorFunc(t, 2 /* update fails twice*/),
+				},
+			},
+			expectedActions: []clienttesting.Action{
+				// Fails
+				clienttesting.NewUpdateAction(pvVer, "", deleted(pv())),
+				// Fails too
+				clienttesting.NewUpdateAction(pvVer, "", deleted(pv())),
+				// Succeeds
+				clienttesting.NewUpdateAction(pvVer, "", deleted(pv())),
+			},
+		},
+		{
+			name:            "deleted PVC with finalizer + PV is bound -> finalizer is not removed",
+			updatedPV:       deleted(withProtectionFinalizer(boundPV())),
+			expectedActions: []clienttesting.Action{},
+		},
+	}
+
+	for _, test := range tests {
+		// Create client with initial data
+		objs := test.initialObjects
+		if test.updatedPV != nil {
+			objs = append(objs, test.updatedPV)
+		}
+
+		client := fake.NewSimpleClientset(objs...)
+
+		// Create informers
+		informers := informers.NewSharedInformerFactory(client, controller.NoResyncPeriodFunc())
+		pvInformer := informers.Core().V1().PersistentVolumes()
+
+		// Populate the informers with initial objects so the controller can
+		// Get() it.
+		for _, obj := range objs {
+			switch obj.(type) {
+			case *v1.PersistentVolume:
+				pvInformer.Informer().GetStore().Add(obj)
+			default:
+				t.Fatalf("Unknown initalObject type: %+v", obj)
+			}
+		}
+
+		// Add reactor to inject test errors.
+		for _, reactor := range test.reactors {
+			client.Fake.PrependReactor(reactor.verb, reactor.resource, reactor.reactorfn)
+		}
+
+		// Create the controller
+		ctrl := NewPVProtectionController(pvInformer, client)
+
+		// Start the test by simulating an event
+		if test.updatedPV != nil {
+			ctrl.pvAddedUpdated(test.updatedPV)
+		}
+
+		// Process the controller queue until we get expected results
+		timeout := time.Now().Add(10 * time.Second)
+		lastReportedActionCount := 0
+		for {
+			if time.Now().After(timeout) {
+				t.Errorf("Test %q: timed out", test.name)
+				break
+			}
+			if ctrl.queue.Len() > 0 {
+				glog.V(5).Infof("Test %q: %d events queue, processing one", test.name, ctrl.queue.Len())
+				ctrl.processNextWorkItem()
+			}
+			if ctrl.queue.Len() > 0 {
+				// There is still some work in the queue, process it now
+				continue
+			}
+			currentActionCount := len(client.Actions())
+			if currentActionCount < len(test.expectedActions) {
+				// Do not log evey wait, only when the action count changes.
+				if lastReportedActionCount < currentActionCount {
+					glog.V(5).Infof("Test %q: got %d actions out of %d, waiting for the rest", test.name, currentActionCount, len(test.expectedActions))
+					lastReportedActionCount = currentActionCount
+				}
+				// The test expected more to happen, wait for the actions.
+				// Most probably it's exponential backoff
+				time.Sleep(10 * time.Millisecond)
+				continue
+			}
+			break
+		}
+		actions := client.Actions()
+
+		if !reflect.DeepEqual(actions, test.expectedActions) {
+			t.Errorf("Test %q: action not expected\nExpected:\n%s\ngot:\n%s", test.name, spew.Sdump(test.expectedActions), spew.Sdump(actions))
+		}
+
+	}
+
+}

pkg/features/kube_features.go

@@ -203,10 +203,10 @@ const (
 	BlockVolume utilfeature.Feature = "BlockVolume"
 
 	// owner: @pospispa
-	//
 	// alpha: v1.9
-	// Postpone deletion of a persistent volume claim in case it is used by a pod
+	//
-	PVCProtection utilfeature.Feature = "PVCProtection"
+	// Postpone deletion of a PV or a PVC when they are being used
+	StorageProtection utilfeature.Feature = "StorageProtection"
 
 	// owner: @aveshagarwal
 	// alpha: v1.9
@@ -267,7 +267,7 @@ var defaultKubernetesFeatureGates = map[utilfeature.Feature]utilfeature.FeatureS
 	CSIPersistentVolume:                         {Default: true, PreRelease: utilfeature.Beta},
 	CustomPodDNS:                                {Default: false, PreRelease: utilfeature.Alpha},
 	BlockVolume:                                 {Default: false, PreRelease: utilfeature.Alpha},
-	PVCProtection:                               {Default: false, PreRelease: utilfeature.Alpha},
+	StorageProtection:                           {Default: false, PreRelease: utilfeature.Alpha},
 	ResourceLimitsPriorityFunction:              {Default: false, PreRelease: utilfeature.Alpha},
 	SupportIPVSProxyMode:                        {Default: false, PreRelease: utilfeature.Beta},
 	SupportPodPidsLimit:                         {Default: false, PreRelease: utilfeature.Alpha},

pkg/kubeapiserver/options/BUILD

@@ -44,7 +44,6 @@ go_library(
         "//plugin/pkg/admission/noderestriction:go_default_library",
         "//plugin/pkg/admission/persistentvolume/label:go_default_library",
         "//plugin/pkg/admission/persistentvolume/resize:go_default_library",
-        "//plugin/pkg/admission/persistentvolumeclaim/pvcprotection:go_default_library",
         "//plugin/pkg/admission/podnodeselector:go_default_library",
         "//plugin/pkg/admission/podpreset:go_default_library",
         "//plugin/pkg/admission/podtolerationrestriction:go_default_library",
@@ -53,6 +52,7 @@ go_library(
         "//plugin/pkg/admission/security/podsecuritypolicy:go_default_library",
         "//plugin/pkg/admission/securitycontext/scdeny:go_default_library",
         "//plugin/pkg/admission/serviceaccount:go_default_library",
+        "//plugin/pkg/admission/storage/storageprotection:go_default_library",
         "//plugin/pkg/admission/storageclass/setdefault:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
         "//vendor/github.com/pborman/uuid:go_default_library",

pkg/kubeapiserver/options/plugins.go

@@ -41,7 +41,6 @@ import (
 	"k8s.io/kubernetes/plugin/pkg/admission/noderestriction"
 	"k8s.io/kubernetes/plugin/pkg/admission/persistentvolume/label"
 	"k8s.io/kubernetes/plugin/pkg/admission/persistentvolume/resize"
-	"k8s.io/kubernetes/plugin/pkg/admission/persistentvolumeclaim/pvcprotection"
 	"k8s.io/kubernetes/plugin/pkg/admission/podnodeselector"
 	"k8s.io/kubernetes/plugin/pkg/admission/podpreset"
 	"k8s.io/kubernetes/plugin/pkg/admission/podtolerationrestriction"
@@ -50,6 +49,7 @@ import (
 	"k8s.io/kubernetes/plugin/pkg/admission/security/podsecuritypolicy"
 	"k8s.io/kubernetes/plugin/pkg/admission/securitycontext/scdeny"
 	"k8s.io/kubernetes/plugin/pkg/admission/serviceaccount"
+	"k8s.io/kubernetes/plugin/pkg/admission/storage/storageprotection"
 	"k8s.io/kubernetes/plugin/pkg/admission/storageclass/setdefault"
 
 	"k8s.io/apimachinery/pkg/util/sets"
@@ -86,7 +86,7 @@ var AllOrderedPlugins = []string{
 	extendedresourcetoleration.PluginName, // ExtendedResourceToleration
 	label.PluginName,                      // PersistentVolumeLabel
 	setdefault.PluginName,                 // DefaultStorageClass
-	pvcprotection.PluginName,              // PVCProtection
+	storageprotection.PluginName,          // StorageProtection
 	gc.PluginName,                         // OwnerReferencesPermissionEnforcement
 	resize.PluginName,                     // PersistentVolumeClaimResize
 	mutatingwebhook.PluginName,            // MutatingAdmissionWebhook
@@ -125,7 +125,7 @@ func RegisterAllAdmissionPlugins(plugins *admission.Plugins) {
 	serviceaccount.Register(plugins)
 	setdefault.Register(plugins)
 	resize.Register(plugins)
-	pvcprotection.Register(plugins)
+	storageprotection.Register(plugins)
 }
 
 // DefaultOffAdmissionPlugins get admission plugins off by default for kube-apiserver.

pkg/kubelet/volumemanager/populator/desired_state_of_world_populator.go

@@ -434,7 +434,7 @@ func (dswp *desiredStateOfWorldPopulator) getPVCExtractPV(
 			err)
 	}
 
-	if utilfeature.DefaultFeatureGate.Enabled(features.PVCProtection) {
+	if utilfeature.DefaultFeatureGate.Enabled(features.StorageProtection) {
 		// Pods that uses a PVC that is being deleted must not be started.
 		//
 		// In case an old kubelet is running without this check or some kubelets

pkg/volume/util/finalizer.go

@@ -19,4 +19,7 @@ package util
 const (
 	// Name of finalizer on PVCs that have a running pod.
 	PVCProtectionFinalizer = "kubernetes.io/pvc-protection"
+
+	// Name of finalizer on PVs that are bound by PVCs
+	PVProtectionFinalizer = "kubernetes.io/pv-protection"
 )

plugin/BUILD

@@ -28,7 +28,6 @@ filegroup(
         "//plugin/pkg/admission/noderestriction:all-srcs",
         "//plugin/pkg/admission/persistentvolume/label:all-srcs",
         "//plugin/pkg/admission/persistentvolume/resize:all-srcs",
-        "//plugin/pkg/admission/persistentvolumeclaim/pvcprotection:all-srcs",
         "//plugin/pkg/admission/podnodeselector:all-srcs",
         "//plugin/pkg/admission/podpreset:all-srcs",
         "//plugin/pkg/admission/podtolerationrestriction:all-srcs",
@@ -37,6 +36,7 @@ filegroup(
         "//plugin/pkg/admission/security:all-srcs",
         "//plugin/pkg/admission/securitycontext/scdeny:all-srcs",
         "//plugin/pkg/admission/serviceaccount:all-srcs",
+        "//plugin/pkg/admission/storage/storageprotection:all-srcs",
         "//plugin/pkg/admission/storageclass/setdefault:all-srcs",
         "//plugin/pkg/auth:all-srcs",
     ],

plugin/pkg/admission/persistentvolumeclaim/pvcprotection/admission.go

@@ -1,111 +0,0 @@
-/*
-Copyright 2017 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 pvcprotection
-
-import (
-	"fmt"
-	"io"
-
-	"github.com/golang/glog"
-
-	admission "k8s.io/apiserver/pkg/admission"
-	"k8s.io/apiserver/pkg/util/feature"
-	api "k8s.io/kubernetes/pkg/apis/core"
-	informers "k8s.io/kubernetes/pkg/client/informers/informers_generated/internalversion"
-	corelisters "k8s.io/kubernetes/pkg/client/listers/core/internalversion"
-	"k8s.io/kubernetes/pkg/features"
-	kubeapiserveradmission "k8s.io/kubernetes/pkg/kubeapiserver/admission"
-	volumeutil "k8s.io/kubernetes/pkg/volume/util"
-)
-
-const (
-	// PluginName is the name of this admission controller plugin
-	PluginName = "PVCProtection"
-)
-
-// Register registers a plugin
-func Register(plugins *admission.Plugins) {
-	plugins.Register(PluginName, func(config io.Reader) (admission.Interface, error) {
-		plugin := newPlugin()
-		return plugin, nil
-	})
-}
-
-// pvcProtectionPlugin holds state for and implements the admission plugin.
-type pvcProtectionPlugin struct {
-	*admission.Handler
-	lister corelisters.PersistentVolumeClaimLister
-}
-
-var _ admission.Interface = &pvcProtectionPlugin{}
-var _ = kubeapiserveradmission.WantsInternalKubeInformerFactory(&pvcProtectionPlugin{})
-
-// newPlugin creates a new admission plugin.
-func newPlugin() *pvcProtectionPlugin {
-	return &pvcProtectionPlugin{
-		Handler: admission.NewHandler(admission.Create),
-	}
-}
-
-func (c *pvcProtectionPlugin) SetInternalKubeInformerFactory(f informers.SharedInformerFactory) {
-	informer := f.Core().InternalVersion().PersistentVolumeClaims()
-	c.lister = informer.Lister()
-	c.SetReadyFunc(informer.Informer().HasSynced)
-}
-
-// ValidateInitialization ensures lister is set.
-func (c *pvcProtectionPlugin) ValidateInitialization() error {
-	if c.lister == nil {
-		return fmt.Errorf("missing lister")
-	}
-	return nil
-}
-
-// Admit sets finalizer on all PVCs. The finalizer is removed by
-// PVCProtectionController when it's not referenced by any pod.
-//
-// This prevents users from deleting a PVC that's used by a running pod.
-func (c *pvcProtectionPlugin) Admit(a admission.Attributes) error {
-	if !feature.DefaultFeatureGate.Enabled(features.PVCProtection) {
-		return nil
-	}
-
-	if a.GetResource().GroupResource() != api.Resource("persistentvolumeclaims") {
-		return nil
-	}
-
-	if len(a.GetSubresource()) != 0 {
-		return nil
-	}
-
-	pvc, ok := a.GetObject().(*api.PersistentVolumeClaim)
-	// if we can't convert then we don't handle this object so just return
-	if !ok {
-		return nil
-	}
-
-	for _, f := range pvc.Finalizers {
-		if f == volumeutil.PVCProtectionFinalizer {
-			// Finalizer is already present, nothing to do
-			return nil
-		}
-	}
-
-	glog.V(4).Infof("adding PVC protection finalizer to %s/%s", pvc.Namespace, pvc.Name)
-	pvc.Finalizers = append(pvc.Finalizers, volumeutil.PVCProtectionFinalizer)
-	return nil
-}

plugin/pkg/admission/storage/storageprotection/BUILD

@@ -3,7 +3,7 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 go_library(
     name = "go_default_library",
     srcs = ["admission.go"],
-    importpath = "k8s.io/kubernetes/plugin/pkg/admission/persistentvolumeclaim/pvcprotection",
+    importpath = "k8s.io/kubernetes/plugin/pkg/admission/storage/storageprotection",
     visibility = ["//visibility:public"],
     deps = [
         "//pkg/apis/core:go_default_library",
@@ -22,7 +22,7 @@ go_test(
     name = "go_default_test",
     srcs = ["admission_test.go"],
     embed = [":go_default_library"],
-    importpath = "k8s.io/kubernetes/plugin/pkg/admission/persistentvolumeclaim/pvcprotection",
+    importpath = "k8s.io/kubernetes/plugin/pkg/admission/storage/storageprotection",
     deps = [
         "//pkg/apis/core:go_default_library",
         "//pkg/client/informers/informers_generated/internalversion:go_default_library",
@@ -31,6 +31,7 @@ go_test(
         "//vendor/github.com/davecgh/go-spew/spew:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/admission:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/util/feature:go_default_library",
     ],

plugin/pkg/admission/storage/storageprotection/admission.go

@@ -0,0 +1,156 @@
+/*
+Copyright 2017 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 storageprotection
+
+import (
+	"fmt"
+	"io"
+
+	"github.com/golang/glog"
+
+	admission "k8s.io/apiserver/pkg/admission"
+	"k8s.io/apiserver/pkg/util/feature"
+	api "k8s.io/kubernetes/pkg/apis/core"
+	informers "k8s.io/kubernetes/pkg/client/informers/informers_generated/internalversion"
+	corelisters "k8s.io/kubernetes/pkg/client/listers/core/internalversion"
+	"k8s.io/kubernetes/pkg/features"
+	kubeapiserveradmission "k8s.io/kubernetes/pkg/kubeapiserver/admission"
+	volumeutil "k8s.io/kubernetes/pkg/volume/util"
+)
+
+const (
+	// PluginName is the name of this admission controller plugin
+	PluginName = "StorageProtection"
+)
+
+// Register registers a plugin
+func Register(plugins *admission.Plugins) {
+	plugins.Register(PluginName, func(config io.Reader) (admission.Interface, error) {
+		plugin := newPlugin()
+		return plugin, nil
+	})
+}
+
+// storageProtectionPlugin holds state for and implements the admission plugin.
+type storageProtectionPlugin struct {
+	*admission.Handler
+
+	pvcLister corelisters.PersistentVolumeClaimLister
+	pvLister  corelisters.PersistentVolumeLister
+}
+
+var _ admission.Interface = &storageProtectionPlugin{}
+var _ = kubeapiserveradmission.WantsInternalKubeInformerFactory(&storageProtectionPlugin{})
+
+// newPlugin creates a new admission plugin.
+func newPlugin() *storageProtectionPlugin {
+	return &storageProtectionPlugin{
+		Handler: admission.NewHandler(admission.Create),
+	}
+}
+
+func (c *storageProtectionPlugin) SetInternalKubeInformerFactory(f informers.SharedInformerFactory) {
+	pvcInformer := f.Core().InternalVersion().PersistentVolumeClaims()
+	c.pvcLister = pvcInformer.Lister()
+	pvInformer := f.Core().InternalVersion().PersistentVolumes()
+	c.pvLister = pvInformer.Lister()
+	c.SetReadyFunc(func() bool {
+		return pvcInformer.Informer().HasSynced() && pvInformer.Informer().HasSynced()
+	})
+}
+
+// ValidateInitialization ensures lister is set.
+func (c *storageProtectionPlugin) ValidateInitialization() error {
+	if c.pvcLister == nil {
+		return fmt.Errorf("missing PVC lister")
+	}
+	if c.pvLister == nil {
+		return fmt.Errorf("missing PV lister")
+	}
+	return nil
+}
+
+var (
+	pvResource  = api.Resource("persistentvolumes")
+	pvcResource = api.Resource("persistentvolumeclaims")
+)
+
+// Admit sets finalizer on all PVCs(PVs). The finalizer is removed by
+// PVCProtectionController(PVProtectionController) when it's not referenced.
+//
+// This prevents users from deleting a PVC that's used by a running pod.
+// This also prevents admin from deleting a PV that's bound by a PVC
+func (c *storageProtectionPlugin) Admit(a admission.Attributes) error {
+	if !feature.DefaultFeatureGate.Enabled(features.StorageProtection) {
+		return nil
+	}
+
+	switch a.GetResource().GroupResource() {
+	case pvResource:
+		return c.admitPV(a)
+	case pvcResource:
+		return c.admitPVC(a)
+
+	default:
+		return nil
+	}
+}
+
+func (c *storageProtectionPlugin) admitPV(a admission.Attributes) error {
+	if len(a.GetSubresource()) != 0 {
+		return nil
+	}
+
+	pv, ok := a.GetObject().(*api.PersistentVolume)
+	// if we can't convert the obj to PV, just return
+	if !ok {
+		return nil
+	}
+	for _, f := range pv.Finalizers {
+		if f == volumeutil.PVProtectionFinalizer {
+			// Finalizer is already present, nothing to do
+			return nil
+		}
+	}
+	glog.V(4).Infof("adding PV protection finalizer to %s", pv.Name)
+	pv.Finalizers = append(pv.Finalizers, volumeutil.PVProtectionFinalizer)
+
+	return nil
+}
+
+func (c *storageProtectionPlugin) admitPVC(a admission.Attributes) error {
+	if len(a.GetSubresource()) != 0 {
+		return nil
+	}
+
+	pvc, ok := a.GetObject().(*api.PersistentVolumeClaim)
+	// if we can't convert the obj to PVC, just return
+	if !ok {
+		return nil
+	}
+
+	for _, f := range pvc.Finalizers {
+		if f == volumeutil.PVCProtectionFinalizer {
+			// Finalizer is already present, nothing to do
+			return nil
+		}
+	}
+
+	glog.V(4).Infof("adding PVC protection finalizer to %s/%s", pvc.Namespace, pvc.Name)
+	pvc.Finalizers = append(pvc.Finalizers, volumeutil.PVCProtectionFinalizer)
+	return nil
+}

plugin/pkg/admission/storage/storageprotection/admission_test.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package pvcprotection
+package storageprotection
 
 import (
 	"fmt"
@@ -25,6 +25,7 @@ import (
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apiserver/pkg/admission"
 	"k8s.io/apiserver/pkg/util/feature"
 	api "k8s.io/kubernetes/pkg/apis/core"
@@ -43,32 +44,76 @@ func TestAdmit(t *testing.T) {
 			Namespace: "ns",
 		},
 	}
+
+	pv := &api.PersistentVolume{
+		TypeMeta: metav1.TypeMeta{
+			Kind: "PersistentVolume",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "pv",
+		},
+	}
 	claimWithFinalizer := claim.DeepCopy()
 	claimWithFinalizer.Finalizers = []string{volumeutil.PVCProtectionFinalizer}
 
+	pvWithFinalizer := pv.DeepCopy()
+	pvWithFinalizer.Finalizers = []string{volumeutil.PVProtectionFinalizer}
+
 	tests := []struct {
 		name           string
+		resource       schema.GroupVersionResource
 		object         runtime.Object
 		expectedObject runtime.Object
 		featureEnabled bool
+		namespace      string
 	}{
 		{
 			"create -> add finalizer",
+			api.SchemeGroupVersion.WithResource("persistentvolumeclaims"),
 			claim,
 			claimWithFinalizer,
 			true,
+			claim.Namespace,
 		},
 		{
 			"finalizer already exists -> no new finalizer",
+			api.SchemeGroupVersion.WithResource("persistentvolumeclaims"),
 			claimWithFinalizer,
 			claimWithFinalizer,
 			true,
+			claimWithFinalizer.Namespace,
 		},
 		{
 			"disabled feature -> no finalizer",
+			api.SchemeGroupVersion.WithResource("persistentvolumeclaims"),
 			claim,
 			claim,
 			false,
+			claim.Namespace,
+		},
+		{
+			"create -> add finalizer",
+			api.SchemeGroupVersion.WithResource("persistentvolumes"),
+			pv,
+			pvWithFinalizer,
+			true,
+			pv.Namespace,
+		},
+		{
+			"finalizer already exists -> no new finalizer",
+			api.SchemeGroupVersion.WithResource("persistentvolumes"),
+			pvWithFinalizer,
+			pvWithFinalizer,
+			true,
+			pvWithFinalizer.Namespace,
+		},
+		{
+			"disabled feature -> no finalizer",
+			api.SchemeGroupVersion.WithResource("persistentvolumes"),
+			pv,
+			pv,
+			false,
+			pv.Namespace,
 		},
 	}
 
@@ -77,15 +122,15 @@ func TestAdmit(t *testing.T) {
 	ctrl.SetInternalKubeInformerFactory(informerFactory)
 
 	for _, test := range tests {
-		feature.DefaultFeatureGate.Set(fmt.Sprintf("PVCProtection=%v", test.featureEnabled))
+		feature.DefaultFeatureGate.Set(fmt.Sprintf("StorageProtection=%v", test.featureEnabled))
 		obj := test.object.DeepCopyObject()
 		attrs := admission.NewAttributesRecord(
 			obj,                  // new object
 			obj.DeepCopyObject(), // old object, copy to be sure it's not modified
-			api.Kind("PersistentVolumeClaim").WithVersion("version"),
+			schema.GroupVersionKind{},
-			claim.Namespace,
+			test.namespace,
-			claim.Name,
+			"foo",
-			api.Resource("persistentvolumeclaims").WithVersion("version"),
+			test.resource,
 			"", // subresource
 			admission.Create,
 			nil, // userInfo
@@ -102,5 +147,5 @@ func TestAdmit(t *testing.T) {
 
 	// Disable the feature for rest of the tests.
 	// TODO: remove after alpha
-	feature.DefaultFeatureGate.Set("PVCProtection=false")
+	feature.DefaultFeatureGate.Set("StorageProtection=false")
 }

plugin/pkg/auth/authorizer/rbac/bootstrappolicy/controller_policy.go

@@ -324,7 +324,7 @@ func buildControllerRoles() ([]rbac.ClusterRole, []rbac.ClusterRoleBinding) {
 			eventsRule(),
 		},
 	})
-	if utilfeature.DefaultFeatureGate.Enabled(features.PVCProtection) {
+	if utilfeature.DefaultFeatureGate.Enabled(features.StorageProtection) {
 		addControllerRole(&controllerRoles, &controllerRoleBindings, rbac.ClusterRole{
 			ObjectMeta: metav1.ObjectMeta{Name: saRolePrefix + "pvc-protection-controller"},
 			Rules: []rbac.PolicyRule{
@@ -334,6 +334,15 @@ func buildControllerRoles() ([]rbac.ClusterRole, []rbac.ClusterRoleBinding) {
 			},
 		})
 	}
+	if utilfeature.DefaultFeatureGate.Enabled(features.StorageProtection) {
+		addControllerRole(&controllerRoles, &controllerRoleBindings, rbac.ClusterRole{
+			ObjectMeta: metav1.ObjectMeta{Name: saRolePrefix + "pv-protection-controller"},
+			Rules: []rbac.PolicyRule{
+				rbac.NewRule("get", "list", "watch", "update").Groups(legacyGroup).Resources("persistentvolumes").RuleOrDie(),
+				eventsRule(),
+			},
+		})
+	}
 
 	return controllerRoles, controllerRoleBindings
 }

test/e2e/storage/pvc_protection.go

@@ -29,7 +29,7 @@ import (
 	"k8s.io/kubernetes/test/e2e/storage/utils"
 )
 
-var _ = utils.SIGDescribe("PVC Protection [Feature:PVCProtection]", func() {
+var _ = utils.SIGDescribe("PVC Protection [Feature:StorageProtection]", func() {
 	var (
 		client                  clientset.Interface
 		nameSpace               string