Add lazy cleanup of PVC on Mysql cluster deletion

pkg/controller/clustercontroller/cleanups.go

@@ -0,0 +1,79 @@
+/*
+Copyright 2018 Platform9, Inc
+
+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 clustercontroller
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/golang/glog"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+)
+
+// CleanupRetryError Indicates retry on error
+type CleanupRetryError struct {
+	msg string
+}
+
+func (e *CleanupRetryError) Error() string {
+	return e.msg
+}
+
+// Cleanup deletes any orphaned objects created by Mysql cluster
+func (c *Controller) Cleanup(ctx context.Context, name string, namespace string) error {
+
+	glog.Infof("Cleaning up cluster: %s", name)
+
+	client := c.k8client
+
+	lbs := map[string]string{
+		"app":           "mysql-operator",
+		"mysql_cluster": name}
+
+	selector := labels.SelectorFromSet(lbs)
+	listOpt := metav1.ListOptions{
+		LabelSelector: selector.String()}
+
+	podList, err := client.CoreV1().Pods(namespace).List(listOpt)
+
+	if err != nil {
+		return fmt.Errorf("listing pods: %s", err)
+	}
+
+	if len(podList.Items) > 0 {
+		msg := fmt.Sprintf("Pods still terminating for cluster: %s", name)
+		glog.V(2).Infof(msg)
+		return &CleanupRetryError{msg: msg}
+	}
+
+	pvcList, err := client.CoreV1().PersistentVolumeClaims(namespace).List(listOpt)
+
+	if err != nil {
+		return fmt.Errorf("listing pvcs: %s", err)
+	}
+
+	for _, pvc := range pvcList.Items {
+		err := client.CoreV1().PersistentVolumeClaims(pvc.Namespace).Delete(pvc.Name, nil)
+		if err != nil {
+			glog.Warningf("Error deleting pvc: %s", err)
+		}
+	}
+
+	return nil
+}

pkg/controller/clustercontroller/controller.go

@@ -48,7 +48,7 @@ import (
 const (
 	initRetryWaitTime = 30 * time.Second
 	workerPeriodTime  = 1 * time.Second
-
+	cleanupRetryTime  = 30 * time.Second
 	// ControllerName is the name of this controller
 	ControllerName = "mysqlclusterController"
 
@@ -72,7 +72,9 @@ type Controller struct {
 	clusterLister     mylisters.MysqlClusterLister
 	podLister         corelisters.PodLister
 
-	queue       workqueue.RateLimitingInterface
+	queue        workqueue.RateLimitingInterface
+	cleanupQueue workqueue.DelayingInterface
+
 	workerWg    sync.WaitGroup
 	syncedFuncs []cache.InformerSynced
 
@@ -122,6 +124,9 @@ func New(ctx *controllerpkg.Context) *Controller {
 	ctrl.podLister = podInformer.Lister()
 	ctrl.syncedFuncs = append(ctrl.syncedFuncs, podInformer.Informer().HasSynced)
 
+	// Cleanup
+	ctrl.cleanupQueue = workqueue.NewNamedDelayingQueue("mysqlcluster-cleanup")
+
 	return ctrl
 
 }
@@ -164,10 +169,15 @@ func (c *Controller) Start(workers int, stopCh <-chan struct{}) error {
 		go wait.Until(func() { c.workerReconcile(stopCh) }, workerPeriodTime, stopCh)
 	}
 
+	// Just one worker for cleanup
+	c.workerWg.Add(1)
+	go wait.Until(func() { c.workerCleanup(stopCh) }, workerPeriodTime, stopCh)
+
 	<-stopCh
 	glog.V(2).Info("Shutting down controller.")
 	c.queue.ShutDown()
 	c.reconcileQueue.ShutDown()
+	c.cleanupQueue.ShutDown()
 	c.cron.Stop()
 	glog.V(2).Info("Wait for workers to exit...")
 	c.workerWg.Wait()
@@ -205,6 +215,7 @@ func (c *Controller) workerController(stopCh <-chan struct{}) {
 			if err != nil && k8errors.IsNotFound(err) {
 				// If the cluster has disappeared, do not re-queue
 				glog.Infof("Removing issuer %q from processing queue", key)
+				c.cleanupQueue.AddAfter(obj, cleanupRetryTime)
 				c.queue.Forget(obj)
 				return nil
 			}
@@ -287,6 +298,59 @@ func (c *Controller) workerReconcile(stopCh <-chan struct{}) {
 	}
 }
 
+func (c *Controller) workerCleanup(stopCh <-chan struct{}) {
+	defer c.workerWg.Done()
+	ctx, cancel := context.WithCancel(context.Background())
+	ctx = util.ContextWithStopCh(ctx, stopCh)
+	defer cancel()
+
+	glog.V(2).Info("Starting cleanup worker.")
+
+	for {
+		obj, shutdown := c.cleanupQueue.Get()
+		if shutdown {
+			break
+		}
+
+		var key string
+		err := func(obj interface{}) error {
+			defer c.cleanupQueue.Done(obj)
+
+			var ok bool
+			if key, ok = obj.(string); !ok {
+				return nil
+			}
+
+			// process items from queue
+			namespace, name, err := cache.SplitMetaNamespaceKey(key)
+			if err != nil {
+				return fmt.Errorf("invalid resource key: %s", key)
+			}
+
+			if err := c.Cleanup(ctx, name, namespace); err != nil {
+				return err
+			}
+
+			return nil
+		}(obj)
+
+		if err != nil {
+			switch err.(type) {
+			case *CleanupRetryError:
+				glog.Infof("%s controller: Re-queuing item %q due to error processing: %s",
+					ControllerName, key, err.Error(),
+				)
+				c.cleanupQueue.AddAfter(obj, cleanupRetryTime)
+				continue
+			default:
+				glog.Errorf("%s controller: Giving up %q due to error: %s",
+					ControllerName, key, err.Error(),
+				)
+			}
+		}
+	}
+}
+
 func (c *Controller) getNextWorkItem(key string) (*api.MysqlCluster, error) {
 	namespace, name, err := cache.SplitMetaNamespaceKey(key)
 	if err != nil {

pkg/mysqlcluster/statefullset.go

@@ -467,19 +467,23 @@ func (f *cFactory) ensureVolumes(in []core.Volume) []core.Volume {
 }
 
 func (f *cFactory) ensureVolumeClaimTemplates(in []core.PersistentVolumeClaim) []core.PersistentVolumeClaim {
-	initPvc := false
+	init := false
 	if len(in) == 0 {
 		in = make([]core.PersistentVolumeClaim, 1)
-		initPvc = true
+		init = true
 	}
+
 	data := in[0]
 
 	data.Name = dataVolumeName
 
-	if initPvc {
-		// This can be set only when creating new PVC. It ensures that PVC can be
-		// terminated after deleting parent MySQL cluster
-		data.ObjectMeta.OwnerReferences = f.getOwnerReferences()
+	if init {
+		if len(data.ObjectMeta.Annotations) == 0 {
+			data.ObjectMeta.Annotations = make(map[string]string)
+		}
+
+		data.ObjectMeta.Annotations["app"] = "mysql-operator"
+		data.ObjectMeta.Annotations["mysql_cluster"] = f.cluster.Name
 	}
 
 	data.Spec = f.cluster.Spec.VolumeSpec.PersistentVolumeClaimSpec