[WIP] feat(PDB, BDD): add test to verify PDB creation

cmd/cvc-operator/controller/cstorvolumeclaim.go

@@ -23,14 +23,17 @@ import (
 
 	apis "github.com/openebs/maya/pkg/apis/openebs.io/v1alpha1"
 	menv "github.com/openebs/maya/pkg/env/v1alpha1"
+	apispdb "github.com/openebs/maya/pkg/kubernetes/poddisruptionbudget"
 	"github.com/openebs/maya/pkg/version"
 
 	cspi "github.com/openebs/maya/pkg/cstor/poolinstance/v1alpha3"
 	cv "github.com/openebs/maya/pkg/cstor/volume/v1alpha1"
 	cvr "github.com/openebs/maya/pkg/cstor/volumereplica/v1alpha1"
+	cvclaim "github.com/openebs/maya/pkg/cstorvolumeclaim/v1alpha1"
 	svc "github.com/openebs/maya/pkg/kubernetes/service/v1alpha1"
 	errors "github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
+	policy "k8s.io/api/policy/v1beta1"
 	k8serror "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
@@ -177,6 +180,11 @@ func getCSPC(
 	return cspcName
 }
 
+// getPDBName returns the PDB name from cStor Volume Claim label
+func getPDBName(claim *apis.CStorVolumeClaim) string {
+	return claim.GetLabels()[string(apis.PodDisruptionBudgetKey)]
+}
+
 // listCStorPools get the list of available pool using the storagePoolClaim
 // as labelSelector.
 func listCStorPools(
@@ -450,7 +458,7 @@ func createCVR(
 func getPoolMapFromCVRList(cvrList *apis.CStorVolumeReplicaList) map[string]bool {
 	var poolMap = make(map[string]bool)
 	for _, cvr := range cvrList.Items {
-		poolName := cvr.GetLabels()[string(cstorpoolInstanceLabel)]
+		poolName := cvr.GetLabels()[string(apis.CstorpoolInstanceLabel)]
 		if poolName != "" {
 			poolMap[poolName] = true
 		}
@@ -521,6 +529,92 @@ func randomizePoolList(list *apis.CStorPoolInstanceList) *apis.CStorPoolInstance
 	for _, randomIdx := range perm {
 		res.Items = append(res.Items, list.Items[randomIdx])
 	}
-
 	return res
 }
+
+// getOrCreatePodDisruptionBudget will does following things
+// 1. It tries to get the PDB that was created among volume replica pools.
+// 2. If PDB exist it returns the PDB.
+// 3. If PDB doesn't exist it creates new PDB(With CSPC hash)
+func getOrCreatePodDisruptionBudget(
+	cvObj *apis.CStorVolume, cspcName string) (*policy.PodDisruptionBudget, error) {
+	poolNames, err := cvr.GetReplicaPoolNames(cvObj)
+	if err != nil {
+		return nil, errors.Wrapf(err,
+			"failed to get volume replica pool names of volume %s",
+			cvObj.Name)
+	}
+	pdbLabels := cvclaim.GetPDBPoolLabels(poolNames)
+	labelSelector := apispdb.GetPDBLabelSelector(pdbLabels)
+	pdbList, err := apispdb.KubeClient().
+		WithNamespace(getNamespace()).
+		List(metav1.ListOptions{LabelSelector: labelSelector})
+	if err != nil {
+		return nil, errors.Wrapf(err,
+			"failed to list PDB belongs to pools %v", pdbLabels)
+	}
+	if len(pdbList.Items) > 1 {
+		return nil, errors.Wrapf(err,
+			"current PDB count %d of pools %v",
+			len(pdbList.Items),
+			pdbLabels)
+	}
+	if len(pdbList.Items) == 1 {
+		return &pdbList.Items[0], nil
+	}
+	return createPDB(cvObj, poolNames, cspcName)
+}
+
+// createPDB creates PDB for cStorVolumes based on arguments
+func createPDB(cvObj *apis.CStorVolume,
+	poolNames []string, cspcName string) (*policy.PodDisruptionBudget, error) {
+	// Calculate minAvailable value from cStorVolume replica count
+	minAvailable := (cvObj.Spec.ReplicationFactor >> 1) + 1
+	minAvailableIntStr := intstr.FromInt(minAvailable)
+
+	//build podDisruptionBudget for volume
+	pdbObj := policy.PodDisruptionBudget{
+		ObjectMeta: metav1.ObjectMeta{
+			GenerateName: cspcName,
+			Labels:       cvclaim.GetPDBLabels(poolNames, cspcName),
+		},
+		Spec: policy.PodDisruptionBudgetSpec{
+			MinAvailable: &minAvailableIntStr,
+			Selector:     getPDBSelector(poolNames),
+		},
+	}
+	// Create podDisruptionBudget
+	return apispdb.KubeClient().
+		WithNamespace(cvObj.Namespace).
+		Create(&pdbObj)
+}
+
+// getPDBSelector returns PDB label selector from list of pools
+func getPDBSelector(pools []string) *metav1.LabelSelector {
+	selectorRequirements := []metav1.LabelSelectorRequirement{}
+	selectorRequirements = append(
+		selectorRequirements,
+		metav1.LabelSelectorRequirement{
+			Key:      string(apis.CStorPoolInstanceCPK),
+			Operator: metav1.LabelSelectorOpIn,
+			Values:   pools,
+		})
+	return &metav1.LabelSelector{
+		MatchLabels: map[string]string{
+			"app": "cstor-pool",
+		},
+		MatchExpressions: selectorRequirements,
+	}
+}
+
+// addPDBLabelOnCVC will add PodDisruptionBudget label on CVC
+func addPDBLabelOnCVC(
+	cvcObj *apis.CStorVolumeClaim, pdbObj *policy.PodDisruptionBudget) *apis.CStorVolumeClaim {
+	cvcLabels := cvcObj.GetLabels()
+	if cvcLabels == nil {
+		cvcLabels = map[string]string{}
+	}
+	cvcLabels[apis.PodDisruptionBudgetKey] = pdbObj.Name
+	cvcObj.SetLabels(cvcLabels)
+	return cvcObj
+}

cmd/cvc-operator/controller/cvc_controller.go

@@ -22,15 +22,20 @@ import (
 	"time"
 
 	apis "github.com/openebs/maya/pkg/apis/openebs.io/v1alpha1"
+	apispdb "github.com/openebs/maya/pkg/kubernetes/poddisruptionbudget"
+	errors "github.com/pkg/errors"
 	merrors "github.com/pkg/errors"
 	"k8s.io/klog"
 
+	cvclaim "github.com/openebs/maya/pkg/cstorvolumeclaim/v1alpha1"
 	corev1 "k8s.io/api/core/v1"
+	policy "k8s.io/api/policy/v1beta1"
 	k8serror "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	klabels "k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
+
 	"k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/strategicpatch"
 	"k8s.io/client-go/kubernetes/scheme"
@@ -63,6 +68,9 @@ const (
 	// CStorVolumeClaimFinalizer name of finalizer on CStorVolumeClaim that
 	// are bound by CStorVolume
 	CStorVolumeClaimFinalizer = "cvc.openebs.io/finalizer"
+	// DeProvisioning is used as part of the event 'reason' during
+	// cstorvolumeclaim deprovisioning stage
+	DeProvisioning = "DeProvisioning"
 )
 
 var knownResizeConditions = map[apis.CStorVolumeClaimConditionType]bool{
@@ -144,7 +152,11 @@ func (c *CVCController) syncCVC(cvc *apis.CStorVolumeClaim) error {
 	// and remove finalizer.
 	if c.isClaimDeletionCandidate(cvc) {
 		klog.Infof("syncClaim: remove finalizer for CStorVolumeClaimVolume [%s]", cvc.Name)
-		return c.removeClaimFinalizer(cvc)
+		err = c.removeClaimFinalizer(cvc)
+		if err != nil {
+			c.recorder.Eventf(cvc, corev1.EventTypeWarning, DeProvisioning, err.Error())
+		}
+		return nil
 	}
 
 	volName := cvc.Name
@@ -193,7 +205,6 @@ func (c *CVCController) syncCVC(cvc *apis.CStorVolumeClaim) error {
 	if err != nil {
 		return err
 	}
-
 	return nil
 }
 
@@ -230,7 +241,9 @@ func (c *CVCController) updateCVCObj(
 // 2. Create cstorvolume resource with required iscsi information.
 // 3. Create target deployment.
 // 4. Create cstorvolumeclaim resource.
-// 5. Update the cstorvolumeclaim with claimRef info and bound with cstorvolume.
+// 5. Create PDB provisioning volume is HA volume.
+// 5. Update the cstorvolumeclaim with claimRef info, PDB label(only for HA
+//    volumes) and bound with cstorvolume.
 func (c *CVCController) createVolumeOperation(cvc *apis.CStorVolumeClaim) (*apis.CStorVolumeClaim, error) {
 	_ = cvc.Annotations[string(apis.ConfigClassKey)]
 
@@ -258,6 +271,16 @@ func (c *CVCController) createVolumeOperation(cvc *apis.CStorVolumeClaim) (*apis
 		return nil, err
 	}
 
+	if cvclaim.IsHAVolume(cvc) {
+		var pdbObj *policy.PodDisruptionBudget
+		pdbObj, err = getOrCreatePodDisruptionBudget(cvObj, getCSPC(cvc))
+		if err != nil {
+			return nil, errors.Wrapf(err,
+				"failed to create PDB for volume: %s", cvc.Name)
+		}
+		cvc = addPDBLabelOnCVC(cvc, pdbObj)
+	}
+
 	volumeRef, err := ref.GetReference(scheme.Scheme, cvObj)
 	if err != nil {
 		return nil, err
@@ -306,6 +329,15 @@ func (c *CVCController) isClaimDeletionCandidate(cvc *apis.CStorVolumeClaim) boo
 func (c *CVCController) removeClaimFinalizer(
 	cvc *apis.CStorVolumeClaim,
 ) error {
+	if cvclaim.IsHAVolume(cvc) {
+		err := c.deletePDBIfNotInUse(cvc)
+		if err != nil {
+			return errors.Wrapf(err,
+				"failed to verify whether PDB %s is in use by other volumes",
+				getPDBName(cvc),
+			)
+		}
+	}
 	cvcPatch := []Patch{
 		Patch{
 			Op:   "remove",
@@ -566,3 +598,27 @@ func (c *CVCController) resizeCV(cv *apis.CStorVolume, newCapacity resource.Quan
 	}
 	return nil
 }
+
+// deletePDBIfNotInUse deletes the PDB if no volume is refering to the
+// cStorvolumeclaim PDB
+func (c *CVCController) deletePDBIfNotInUse(cvc *apis.CStorVolumeClaim) error {
+	pdbName := getPDBName(cvc)
+	cvcLabelSelector := string(apis.PodDisruptionBudgetKey) + "=" + pdbName
+	cvcList, err := c.clientset.
+		OpenebsV1alpha1().
+		CStorVolumeClaims(cvc.Namespace).
+		List(metav1.ListOptions{LabelSelector: cvcLabelSelector})
+	if err != nil {
+		return errors.Wrapf(err,
+			"failed to list volumes refering to PDB %s", pdbName)
+	}
+	if len(cvcList.Items) == 1 {
+		err = apispdb.KubeClient().
+			WithNamespace(cvc.Namespace).
+			Delete(pdbName, &metav1.DeleteOptions{})
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}

pkg/apis/openebs.io/v1alpha1/cas_keys.go

@@ -85,6 +85,12 @@ const (
 	// PredecessorBDKey is the key to fetch the predecessor BD in case of
 	// block device replacement.
 	PredecessorBDKey = "openebs.io/bd-predecessor"
+
+	//PodDisruptionBudgetKey is the key used to identify the PDB
+	PodDisruptionBudgetKey = "openebs.io/pod-disruption-budget"
+
+	// CstorpoolInstanceLabel is the key used on pool dependent resources
+	CstorpoolInstanceLabel = "cstorpoolinstance.openebs.io/name"
 )
 
 // CASPlainKey represents a openebs key used either in resource annotation

pkg/apis/openebs.io/v1alpha1/cas_pool.go

@@ -37,6 +37,11 @@ const (
 	StoragePoolClaimCPK CasPoolKey = "openebs.io/storage-pool-claim"
 	// CStorPoolClusterCPK is the CStorPoolcluster label
 	CStorPoolClusterCPK CasPoolKey = "openebs.io/cstor-pool-cluster"
+	// CStorPoolInstanceCPK is the CStorPoolInstance label
+	CStorPoolInstanceCPK CasPoolKey = "openebs.io/cstor-pool-instance"
+	// PredecessorBlockDeviceCPK is the annotation on the block device claim
+	// holding previous block device name
+	PredecessorBlockDeviceCPK CasPoolKey = "openebs.io/bd-predecessor"
 	// NdmDiskTypeCPK is the node-disk-manager disk type e.g. 'sparse' or 'disk'
 	NdmDiskTypeCPK CasPoolKey = "ndm.io/disk-type"
 	// NdmBlockDeviceTypeCPK is the node-disk-manager blockdevice type e.g. // 'blockdevice'
@@ -63,6 +68,9 @@ const (
 	RaidzBlockDeviceCountCPV CasPoolValInt = 3
 	// Raidz2BlockDeviceCountCPV is the count for raidz2 type pool
 	Raidz2BlockDeviceCountCPV CasPoolValInt = 6
+	// PersistentVolumeCPK is a key on the various resources to identify it belongs
+	// to particular volume
+	PersistentVolumeCPK CasPoolKey = "openebs.io/persistent-volume"
 )
 
 // CasPool is a type which will be utilised by CAS engine to perform

pkg/apis/openebs.io/v1alpha1/cstor_volume_claim.go

@@ -59,6 +59,12 @@ type CStorVolumeClaimSpec struct {
 	// CstorVolumeSource contains the source volumeName@snapShotname
 	// combaination.  This will be filled only if it is a clone creation.
 	CstorVolumeSource string `json:"cstorVolumeSource,omitempty"`
+	// PodDisruptionBudget will handle the podDisruptionBudget of volume
+	// 1. If podDisruptionBudget is set true then CVC controller will create PDB with
+	//    minAvailable value as 51% of replica count.
+	// 2. If podDisruptionBudget is set false then CVC controller will delete if there
+	//    is any existing PDB.
+	PodDisruptionBudget bool `json:"podDisruptionBudget,omitempty"`
 }
 
 // CStorVolumeClaimPublish contains info related to attachment of a volume to a node.

pkg/cstor/volumereplica/v1alpha1/utils.go

@@ -0,0 +1,55 @@
+/*
+Copyright 2019 The OpenEBS 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 v1alpha1
+
+import (
+	apis "github.com/openebs/maya/pkg/apis/openebs.io/v1alpha1"
+	errors "github.com/pkg/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// getCVRList returns list of volume replicas related to provided volume
+func getCVRList(cvObj *apis.CStorVolume) (*apis.CStorVolumeReplicaList, error) {
+	pvName := cvObj.Labels[string(apis.PersistentVolumeCPK)]
+	pvLabel := string(apis.PersistentVolumeCPK) + "=" + pvName
+	return NewKubeclient(WithNamespace(cvObj.Namespace)).
+		List(metav1.ListOptions{
+			LabelSelector: pvLabel,
+		})
+}
+
+// getPoolNames returns list of pool names from cStor volume replcia list
+func getPoolNames(cvrList *apis.CStorVolumeReplicaList) []string {
+	poolNames := []string{}
+	for _, cvrObj := range cvrList.Items {
+		poolNames = append(poolNames, cvrObj.Labels[string(apis.CstorpoolInstanceLabel)])
+	}
+	return poolNames
+}
+
+// GetReplicaPoolNames return list of replicas pool names by taking cStor
+// volume claim as a argument and return error(if any error occured)
+func GetReplicaPoolNames(cvObj *apis.CStorVolume) ([]string, error) {
+	pvName := cvObj.Labels[string(apis.PersistentVolumeCPK)]
+	cvrList, err := getCVRList(cvObj)
+	if err != nil {
+		return []string{}, errors.Wrapf(err,
+			"failed to list cStorVolumeReplicas related to volume %s",
+			pvName)
+	}
+	return getPoolNames(cvrList), nil
+}