Added PVC Protection Controller

This controller removes protection finalizer from PVCs that are being
deleted and are not referenced by any pod.

cmd/kube-controller-manager/app/BUILD

@@ -79,6 +79,7 @@ go_library(
         "//pkg/controller/volume/attachdetach:go_default_library",
         "//pkg/controller/volume/expand:go_default_library",
         "//pkg/controller/volume/persistentvolume:go_default_library",
+        "//pkg/controller/volume/pvcprotection: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

@@ -359,6 +359,7 @@ func NewControllerInitializers() map[string]InitFunc {
 	controllers["attachdetach"] = startAttachDetachController
 	controllers["persistentvolume-expander"] = startVolumeExpandController
 	controllers["clusterrole-aggregation"] = startClusterRoleAggregrationController
+	controllers["pvc-protection"] = startPVCProtectionController
 
 	return controllers
 }

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

@@ -53,6 +53,7 @@ import (
 	"k8s.io/kubernetes/pkg/controller/volume/attachdetach"
 	"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/features"
 	"k8s.io/kubernetes/pkg/quota/generic"
 	quotainstall "k8s.io/kubernetes/pkg/quota/install"
@@ -376,3 +377,15 @@ func startGarbageCollectorController(ctx ControllerContext) (bool, error) {
 
 	return true, nil
 }
+
+func startPVCProtectionController(ctx ControllerContext) (bool, error) {
+	if utilfeature.DefaultFeatureGate.Enabled(features.PVCProtection) {
+		go pvcprotection.NewPVCProtectionController(
+			ctx.InformerFactory.Core().V1().PersistentVolumeClaims(),
+			ctx.InformerFactory.Core().V1().Pods(),
+			ctx.ClientBuilder.ClientOrDie("pvc-protection-controller"),
+		).Run(1, ctx.Stop)
+		return true, nil
+	}
+	return false, nil
+}

pkg/controller/BUILD

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

pkg/controller/volume/pvcprotection/BUILD

@@ -0,0 +1,61 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["pvc_protection_controller.go"],
+    importpath = "k8s.io/kubernetes/pkg/controller/volume/pvcprotection",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/controller:go_default_library",
+        "//pkg/util/metrics:go_default_library",
+        "//pkg/volume/util:go_default_library",
+        "//pkg/volume/util/volumehelper: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/labels: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",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["pvc_protection_controller_test.go"],
+    importpath = "k8s.io/kubernetes/pkg/controller/volume/pvcprotection",
+    library = ":go_default_library",
+    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",
+    ],
+)
+
+filegroup(
+    name = "package-srcs",
+    srcs = glob(["**"]),
+    tags = ["automanaged"],
+    visibility = ["//visibility:private"],
+)
+
+filegroup(
+    name = "all-srcs",
+    srcs = [":package-srcs"],
+    tags = ["automanaged"],
+    visibility = ["//visibility:public"],
+)

pkg/controller/volume/pvcprotection/pvc_protection_controller.go

@@ -0,0 +1,284 @@
+/*
+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"
+	"time"
+
+	"github.com/golang/glog"
+	"k8s.io/api/core/v1"
+	apierrs "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/labels"
+	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"
+	volumeutil "k8s.io/kubernetes/pkg/volume/util"
+	"k8s.io/kubernetes/pkg/volume/util/volumehelper"
+)
+
+// Controller is controller that removes PVCProtectionFinalizer
+// from PVCs that are used by no pods.
+type Controller struct {
+	client clientset.Interface
+
+	pvcLister       corelisters.PersistentVolumeClaimLister
+	pvcListerSynced cache.InformerSynced
+
+	podLister       corelisters.PodLister
+	podListerSynced cache.InformerSynced
+
+	queue workqueue.RateLimitingInterface
+}
+
+// NewPVCProtectionController returns a new *{VCProtectionController.
+func NewPVCProtectionController(pvcInformer coreinformers.PersistentVolumeClaimInformer, podInformer coreinformers.PodInformer, cl clientset.Interface) *Controller {
+	e := &Controller{
+		client: cl,
+		queue:  workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "pvcprotection"),
+	}
+	if cl != nil && cl.CoreV1().RESTClient().GetRateLimiter() != nil {
+		metrics.RegisterMetricAndTrackRateLimiterUsage("persistentvolumeclaim_protection_controller", cl.CoreV1().RESTClient().GetRateLimiter())
+	}
+
+	e.pvcLister = pvcInformer.Lister()
+	e.pvcListerSynced = pvcInformer.Informer().HasSynced
+	pvcInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: e.pvcAddedUpdated,
+		UpdateFunc: func(old, new interface{}) {
+			e.pvcAddedUpdated(new)
+		},
+	})
+
+	e.podLister = podInformer.Lister()
+	e.podListerSynced = podInformer.Informer().HasSynced
+	podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			e.podAddedDeletedUpdated(obj, false)
+		},
+		DeleteFunc: func(obj interface{}) {
+			e.podAddedDeletedUpdated(obj, true)
+		},
+		UpdateFunc: func(old, new interface{}) {
+			e.podAddedDeletedUpdated(new, false)
+		},
+	})
+
+	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 PVC protection controller")
+	defer glog.Infof("Shutting down PVC protection controller")
+
+	if !controller.WaitForCacheSync("PVC protection", stopCh, c.pvcListerSynced, c.podListerSynced) {
+		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 {
+	pvcKey, quit := c.queue.Get()
+	if quit {
+		return false
+	}
+	defer c.queue.Done(pvcKey)
+
+	pvcNamespace, pvcName, err := cache.SplitMetaNamespaceKey(pvcKey.(string))
+	if err != nil {
+		utilruntime.HandleError(fmt.Errorf("Error parsing PVC key %q: %v", pvcKey, err))
+		return true
+	}
+
+	err = c.processPVC(pvcNamespace, pvcName)
+	if err == nil {
+		c.queue.Forget(pvcKey)
+		return true
+	}
+
+	utilruntime.HandleError(fmt.Errorf("PVC %v failed with : %v", pvcKey, err))
+	c.queue.AddRateLimited(pvcKey)
+
+	return true
+}
+
+func (c *Controller) processPVC(pvcNamespace, pvcName string) error {
+	glog.V(4).Infof("Processing PVC %s/%s", pvcNamespace, pvcName)
+	startTime := time.Now()
+	defer func() {
+		glog.V(4).Infof("Finished processing PVC %s/%s (%v)", pvcNamespace, pvcName, time.Now().Sub(startTime))
+	}()
+
+	pvc, err := c.pvcLister.PersistentVolumeClaims(pvcNamespace).Get(pvcName)
+	if apierrs.IsNotFound(err) {
+		glog.V(4).Infof("PVC %s/%s not found, ignoring", pvcNamespace, pvcName)
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+
+	if volumeutil.IsPVCBeingDeleted(pvc) && volumeutil.IsProtectionFinalizerPresent(pvc) {
+		// PVC should be deleted. Check if it's used and remove finalizer if
+		// it's not.
+		isUsed, err := c.isBeingUsed(pvc)
+		if err != nil {
+			return err
+		}
+		if !isUsed {
+			return c.removeFinalizer(pvc)
+		}
+	}
+
+	if !volumeutil.IsPVCBeingDeleted(pvc) && !volumeutil.IsProtectionFinalizerPresent(pvc) {
+		// PVC 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 PVCs that were created before the admission
+		// plugin was enabled.
+		return c.addFinalizer(pvc)
+	}
+	return nil
+}
+
+func (c *Controller) addFinalizer(pvc *v1.PersistentVolumeClaim) error {
+	claimClone := pvc.DeepCopy()
+	volumeutil.AddProtectionFinalizer(claimClone)
+	_, err := c.client.CoreV1().PersistentVolumeClaims(claimClone.Namespace).Update(claimClone)
+	if err != nil {
+		glog.V(3).Infof("Error adding protection finalizer to PVC %s/%s: %v", pvc.Namespace, pvc.Name)
+		return err
+	}
+	glog.V(3).Infof("Added protection finalizer to PVC %s/%s", pvc.Namespace, pvc.Name)
+	return nil
+}
+
+func (c *Controller) removeFinalizer(pvc *v1.PersistentVolumeClaim) error {
+	claimClone := pvc.DeepCopy()
+	volumeutil.RemoveProtectionFinalizer(claimClone)
+	_, err := c.client.CoreV1().PersistentVolumeClaims(claimClone.Namespace).Update(claimClone)
+	if err != nil {
+		glog.V(3).Infof("Error removing protection finalizer from PVC %s/%s: %v", pvc.Namespace, pvc.Name, err)
+		return err
+	}
+	glog.V(3).Infof("Removed protection finalizer from PVC %s/%s", pvc.Namespace, pvc.Name)
+	return nil
+}
+
+func (c *Controller) isBeingUsed(pvc *v1.PersistentVolumeClaim) (bool, error) {
+	pods, err := c.podLister.Pods(pvc.Namespace).List(labels.Everything())
+	if err != nil {
+		return false, err
+	}
+	for _, pod := range pods {
+		if pod.Spec.NodeName == "" {
+			// This pod is not scheduled. We have a predicated in scheduler that
+			// prevents scheduling pods with deletion timestamp, so we can be
+			// pretty sure it won't be scheduled in parallel to this check.
+			// Therefore this pod does not block the PVC from deletion.
+			glog.V(4).Infof("Skipping unscheduled pod %s when checking PVC %s/%s", pod.Name, pvc.Namespace, pvc.Name)
+			continue
+		}
+		if volumehelper.IsPodTerminated(pod, pod.Status) {
+			// This pod is being unmounted/detached or is already
+			// unmounted/detached. It does not block the PVC from deletion.
+			continue
+		}
+		for _, volume := range pod.Spec.Volumes {
+			if volume.PersistentVolumeClaim == nil {
+				continue
+			}
+			if volume.PersistentVolumeClaim.ClaimName == pvc.Name {
+				glog.V(2).Infof("Keeping PVC %s/%s, it is used by pod %s/%s", pvc.Namespace, pvc.Name, pod.Namespace, pod.Name)
+				return true, nil
+			}
+		}
+	}
+
+	glog.V(3).Infof("PVC %s/%s is unused", pvc.Namespace, pvc.Name)
+	return false, nil
+}
+
+// pvcAddedUpdated reacts to pvc added/updated/deleted events
+func (c *Controller) pvcAddedUpdated(obj interface{}) {
+	pvc, ok := obj.(*v1.PersistentVolumeClaim)
+	if !ok {
+		utilruntime.HandleError(fmt.Errorf("PVC informer returned non-PVC object: %#v", obj))
+		return
+	}
+	key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(pvc)
+	if err != nil {
+		utilruntime.HandleError(fmt.Errorf("Couldn't get key for Persistent Volume Claim %#v: %v", pvc, err))
+		return
+	}
+	glog.V(4).Infof("Got event on PVC %s", key)
+
+	if (!volumeutil.IsPVCBeingDeleted(pvc) && !volumeutil.IsProtectionFinalizerPresent(pvc)) || (volumeutil.IsPVCBeingDeleted(pvc) && volumeutil.IsProtectionFinalizerPresent(pvc)) {
+		c.queue.Add(key)
+	}
+}
+
+// podAddedDeletedUpdated reacts to Pod events
+func (c *Controller) podAddedDeletedUpdated(obj interface{}, deleted bool) {
+	pod, ok := obj.(*v1.Pod)
+	if !ok {
+		tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
+		if !ok {
+			utilruntime.HandleError(fmt.Errorf("Couldn't get object from tombstone %#v", obj))
+			return
+		}
+		pod, ok = tombstone.Obj.(*v1.Pod)
+		if !ok {
+			utilruntime.HandleError(fmt.Errorf("Tombstone contained object that is not a Pod %#v", obj))
+			return
+		}
+	}
+
+	// Filter out pods that can't help us to remove a finalizer on PVC
+	if !deleted && !volumehelper.IsPodTerminated(pod, pod.Status) && pod.Spec.NodeName != "" {
+		return
+	}
+
+	glog.V(4).Infof("Got event on pod %s/%s", pod.Namespace, pod.Name)
+
+	// Enqueue all PVCs that the pod uses
+	for _, volume := range pod.Spec.Volumes {
+		if volume.PersistentVolumeClaim != nil {
+			c.queue.Add(pod.Namespace + "/" + volume.PersistentVolumeClaim.ClaimName)
+		}
+	}
+}