feat(cvc-operator): add automatic scaling of volumereplicas for CSI volumes

cmd/cvc-operator/controller/cvc_controller.go

@@ -26,6 +26,8 @@ import (
 	errors "github.com/pkg/errors"
 	"k8s.io/klog"
 
+	cvr "github.com/openebs/maya/pkg/cstor/volumereplica/v1alpha1"
+	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"
@@ -203,6 +205,12 @@ func (c *CVCController) syncCVC(cvc *apis.CStorVolumeClaim) error {
 	if err != nil {
 		return err
 	}
+
+	if c.isCVCScalePending(cvc) {
+		// process scale-up/scale-down of volume replicas only if there is
+		// change in curent and desired state of replicas pool information
+		_ = c.scaleVolumeReplicas(cvc)
+	}
 	return nil
 }
 
@@ -274,11 +282,19 @@ func (c *CVCController) createVolumeOperation(cvc *apis.CStorVolumeClaim) (*apis
 		return nil, err
 	}
 
+	// Fetch the volume replica pool names and use them in PDB and updating in
+	// spec and status of CVC
+	poolNames, err := cvr.GetVolumeReplicaPoolNames(cvc.Name, openebsNamespace)
+	if err != nil {
+		return nil, errors.Wrapf(err,
+			"failed to get volume replica pool names of volume %s", cvObj.Name)
+	}
+
 	if isHAVolume(cvc) {
 		// TODO: When multiple threads or multiple CVC controllers are set then
 		// we have to revist entier PDB code path
 		var pdbObj *policy.PodDisruptionBudget
-		pdbObj, err = getOrCreatePodDisruptionBudget(cvObj, getCSPC(cvc))
+		pdbObj, err = getOrCreatePodDisruptionBudget(getCSPC(cvc), poolNames)
 		if err != nil {
 			return nil, errors.Wrapf(err,
 				"failed to create PDB for volume: %s", cvc.Name)
@@ -298,6 +314,13 @@ func (c *CVCController) createVolumeOperation(cvc *apis.CStorVolumeClaim) (*apis
 	cvc.Status.Phase = apis.CStorVolumeClaimPhaseBound
 	cvc.Status.Capacity = cvc.Spec.Capacity
 
+	// TODO: Below function needs to be converted into
+	// cvc.addReplicaPoolInfo(poolNames) while moving to cstor-operators
+	// repo(Currently in Maya writing functions in API package is not encouraged)
+
+	// update volume replica pool information on cvc spec and status
+	addReplicaPoolInfo(cvc, poolNames)
+
 	err = c.updateCVCObj(cvc, cvObj)
 	if err != nil {
 		return nil, err
@@ -315,7 +338,7 @@ func (c *CVCController) getVolumePolicy(
 
 	if policyName != "" {
 		klog.Infof("uses cstorvolume policy %q to configure volume %q", policyName, cvc.Name)
-		volumePolicy, err = c.clientset.OpenebsV1alpha1().CStorVolumePolicies(getNamespace()).Get(policyName, metav1.GetOptions{})
+		volumePolicy, err = c.clientset.OpenebsV1alpha1().CStorVolumePolicies(openebsNamespace).Get(policyName, metav1.GetOptions{})
 		if err != nil {
 			return nil, errors.Wrapf(
 				err,
@@ -335,7 +358,8 @@ func (c *CVCController) isReplicaAffinityEnabled(policy *apis.CStorVolumePolicy)
 }
 
 // distributePendingCVRs trigers create and distribute pending cstorvolumereplica
-// resource among the available cstor pools
+// resource among the available cstor pools. This func returns error even when
+// required no.of CVRs are Not created
 func (c *CVCController) distributePendingCVRs(
 	cvc *apis.CStorVolumeClaim,
 	cv *apis.CStorVolume,
@@ -366,7 +390,7 @@ func (c *CVCController) removeClaimFinalizer(
 	cvc *apis.CStorVolumeClaim,
 ) error {
 	if isHAVolume(cvc) {
-		err := c.deletePDBIfNotInUse(cvc)
+		err := deletePDBIfNotInUse(cvc)
 		if err != nil {
 			return errors.Wrapf(err,
 				"failed to verify whether PDB %s is in use by other volumes",
@@ -627,7 +651,7 @@ func (c *CVCController) resizeCV(cv *apis.CStorVolume, newCapacity resource.Quan
 	if err != nil {
 		return fmt.Errorf("can't update capacity of CV %s as generate patch data failed: %v", cv.Name, err)
 	}
-	_, updateErr := c.clientset.OpenebsV1alpha1().CStorVolumes(getNamespace()).
+	_, updateErr := c.clientset.OpenebsV1alpha1().CStorVolumes(openebsNamespace).
 		Patch(cv.Name, types.MergePatchType, patchBytes)
 	if updateErr != nil {
 		return updateErr
@@ -637,14 +661,13 @@ func (c *CVCController) resizeCV(cv *apis.CStorVolume, newCapacity resource.Quan
 
 // deletePDBIfNotInUse deletes the PDB if no volume is refering to the
 // cStorvolumeclaim PDB
-func (c *CVCController) deletePDBIfNotInUse(cvc *apis.CStorVolumeClaim) error {
+func deletePDBIfNotInUse(cvc *apis.CStorVolumeClaim) error {
 	//TODO: If HALease is enabled active-active then below code needs to be
 	//revist
 	pdbName := getPDBName(cvc)
 	cvcLabelSelector := string(apis.PodDisruptionBudgetKey) + "=" + pdbName
-	cvcList, err := c.clientset.
-		OpenebsV1alpha1().
-		CStorVolumeClaims(cvc.Namespace).
+	cvcList, err := cvclaim.NewKubeclient().
+		WithNamespace(cvc.Namespace).
 		List(metav1.ListOptions{LabelSelector: cvcLabelSelector})
 	if err != nil {
 		return errors.Wrapf(err,
@@ -654,9 +677,44 @@ func (c *CVCController) deletePDBIfNotInUse(cvc *apis.CStorVolumeClaim) error {
 		err = apispdb.KubeClient().
 			WithNamespace(openebsNamespace).
 			Delete(pdbName, &metav1.DeleteOptions{})
+		if k8serror.IsNotFound(err) {
+			klog.Infof("pdb %s of volume %s was already deleted", pdbName, cvc.Name)
+			return nil
+		}
 		if err != nil {
 			return err
 		}
+		klog.Infof("Successfully deleted the PDB %s of volume %s", pdbName, cvc.Name)
+	}
+	return nil
+}
+
+// scaleVolumeReplicas identifies whether it is scaleup or scaledown case of
+// volume replicas. If user added entry of pool info under the spec then changes
+// are treated as scaleup case. If user removed poolInfo entry from spec then
+// changes are treated as scale down case. If user just modifies the pool entry
+// info under the spec then it is a kind of migration which is not yet supported
+func (c *CVCController) scaleVolumeReplicas(cvc *apis.CStorVolumeClaim) error {
+	var err error
+	if len(cvc.Spec.Policy.ReplicaPoolInfo) > len(cvc.Status.PoolInfo) {
+		cvc, err = scaleUpVolumeReplicas(cvc)
+	} else if len(cvc.Spec.Policy.ReplicaPoolInfo) < len(cvc.Status.PoolInfo) {
+		cvc, err = scaleDownVolumeReplicas(cvc)
+	} else {
+		c.recorder.Event(cvc, corev1.EventTypeWarning, "Migration",
+			"Migration of volume replicas is not yet supported")
+		return nil
+	}
+	if err != nil {
+		c.recorder.Eventf(cvc,
+			corev1.EventTypeWarning,
+			"ScalingVolumeReplicas",
+			"%v", err)
+		return err
 	}
+	c.recorder.Eventf(cvc,
+		corev1.EventTypeNormal,
+		"ScalingVolumeReplicas",
+		"successfully scaled volume replicas to %d", len(cvc.Status.PoolInfo))
 	return nil
 }

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

@@ -95,6 +95,8 @@ type CStorVolumeClaimStatus struct {
 	// Capacity the actual resources of the underlying volume.
 	Capacity   corev1.ResourceList         `json:"capacity,omitempty"`
 	Conditions []CStorVolumeClaimCondition `json:"condition,omitempty"`
+	// PoolInfo represents current pool names where volume replicas exists
+	PoolInfo []string `json:"poolInfo"`
 }
 
 // CStorVolumeClaimCondition contains details about state of cstor volume

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

@@ -45,6 +45,9 @@ type CStorVolumePolicySpec struct {
 	Target TargetSpec `json:"target"`
 	// ReplicaSpec represents configuration related to replicas resources
 	Replica ReplicaSpec `json:"replica"`
+	// ReplicaPoolInfo holds the pool information of volume replicas.
+	// Ex: If volume is provisioned on which CStor pool volume replicas exist
+	ReplicaPoolInfo []ReplicaPoolInfo `json:"replicaPoolInfo"`
 }
 
 // TargetSpec represents configuration related to cstor target and its resources
@@ -102,6 +105,13 @@ type Provision struct {
 	ReplicaAffinity bool `json:"replicaAffinity"`
 }
 
+// ReplicaPoolInfo represents the pool information of volume replica
+type ReplicaPoolInfo struct {
+	// PoolName represents the pool name where volume replica exists
+	PoolName string `json:"poolName"`
+	// UID also can be added
+}
+
 // CStorVolumePolicyStatus is for handling status of CstorVolumePolicy
 type CStorVolumePolicyStatus struct {
 	Phase string `json:"phase"`

pkg/cstor/volume/v1alpha1/cstorvolume.go

@@ -448,3 +448,10 @@ func (csr *CVReplicationDetails) UpdateCVWithReplicationDetails(kubeclient *Kube
 	}
 	return err
 }
+
+// IsScaleDownInProgress return true if length of status replica details is
+// greater than length of spec replica details
+func IsScaleDownInProgress(cvObj *apis.CStorVolume) bool {
+	return len(cvObj.Status.ReplicaDetails.KnownReplicas) >
+		len(cvObj.Spec.ReplicaDetails.KnownReplicas)
+}

pkg/cstor/volumereplica/v1alpha1/build.go

@@ -204,8 +204,8 @@ func (b *Builder) WithCapacity(capacity string) *Builder {
 
 // WithStatusPhase sets the Status Phase of CStorVolumeReplica with provided
 //arguments
-func (b *Builder) WithStatusPhase(phase string) *Builder {
-	b.cvr.object.Status.Phase = apis.CStorVolumeReplicaPhase(phase)
+func (b *Builder) WithStatusPhase(phase apis.CStorVolumeReplicaPhase) *Builder {
+	b.cvr.object.Status.Phase = phase
 	return b
 }
 

pkg/cstor/volumereplica/v1alpha1/cstorvolumereplica.go

@@ -111,6 +111,13 @@ func (b *ListBuilder) WithAPIList(
 	return b
 }
 
+// AppendListBuilder append the provided CVR API object into existing ListBuilder
+func (b *ListBuilder) AppendListBuilder(
+	cvr *apis.CStorVolumeReplica) *ListBuilder {
+	b.list.items = append(b.list.items, &CVR{object: cvr})
+	return b
+}
+
 // List returns the list of cvr
 // instances that was built by this
 // builder
@@ -145,6 +152,20 @@ func (p *CVR) IsHealthy() bool {
 	return p.object.Status.Phase == "Healthy"
 }
 
+// HasLabel returns true only if label key value matched to provided
+// value.
+func (p *CVR) HasLabel(key, value string) bool {
+	return p.object.Labels[key] == value
+}
+
+// HasLabel returns predicate to filter out CVRs based on the
+// provided key and values
+func HasLabel(key, value string) Predicate {
+	return func(p *CVR) bool {
+		return p.HasLabel(key, value)
+	}
+}
+
 // IsHealthy is a Predicate to filter out cvrs
 // which is healthy
 func IsHealthy() Predicate {

pkg/cstorvolumeclaim/v1alpha1/utils.go

@@ -61,3 +61,12 @@ func GetPDBLabels(poolNames []string, cspcName string) map[string]string {
 	pdbLabels[string(apis.CStorPoolClusterCPK)] = cspcName
 	return pdbLabels
 }
+
+// GetDesiredReplicaPoolNames returns list of desired pool names
+func GetDesiredReplicaPoolNames(cvc *apis.CStorVolumeClaim) []string {
+	poolNames := []string{}
+	for _, poolInfo := range cvc.Spec.Policy.ReplicaPoolInfo {
+		poolNames = append(poolNames, poolInfo.PoolName)
+	}
+	return poolNames
+}

pkg/util/util.go

@@ -218,3 +218,26 @@ func RemoveString(slice []string, s string) (result []string) {
 	}
 	return result
 }
+
+// IsChangeInLists returns true if there is any difference in listA and listB
+func IsChangeInLists(listA, listB []string) bool {
+	listAMap := map[string]bool{}
+	listBMap := map[string]bool{}
+	for _, name := range listA {
+		listAMap[name] = true
+	}
+	for _, name := range listB {
+		listBMap[name] = true
+	}
+	for _, name := range listA {
+		if !listBMap[name] {
+			return true
+		}
+	}
+	for _, name := range listB {
+		if !listAMap[name] {
+			return true
+		}
+	}
+	return false
+}

tests/cstor/volume/replica_replace/replica_replace_utils_test.go

@@ -206,7 +206,7 @@ func buildCVRFromExistingCVR(
 		WithCapacity(cvrObj.Spec.Capacity).
 		WithReplicaID(cvrObj.Spec.ReplicaID).
 		WithFinalizers([]string{cvr.CStorVolumeReplicaFinalizer}).
-		WithStatusPhase("Recreate")
+		WithStatusPhase(apis.CVRStatusRecreate)
 	cvrObj, err := buildCVRObj.Build()
 	Expect(err).To(BeNil())
 	return cvrObj

tests/operations.go

@@ -1145,7 +1145,7 @@ func (ops *Operations) BuildAndCreateCVR() *apis.CStorVolumeReplica {
 		WithFinalizers([]string{cvr.CStorVolumeReplicaFinalizer}).
 		WithCapacity(cvrConfig.Capacity).
 		WithTargetIP(cvrConfig.TargetIP).
-		WithStatusPhase(cvrConfig.Phase).
+		WithStatusPhase(apis.CStorVolumeReplicaPhase(cvrConfig.Phase)).
 		WithReplicaID(cvrConfig.ReplicaID).
 		Build()
 	Expect(err).To(BeNil())