feat(spc,csp): adding support for pool ReadOnly Threshold limit (#1609)

Signed-off-by: mayank <[email protected]>

cmd/cstor-pool-mgmt/controller/common/common.go

@@ -97,6 +97,10 @@ const (
 	MessageResourceAlreadyPresent EventReason = "Resource already present"
 	// MessageImproperPoolStatus holds message for corresponding failed validate resource.
 	MessageImproperPoolStatus EventReason = "Improper pool status"
+	// PoolROThreshold holds status for pool read only state
+	PoolROThreshold EventReason = "PoolReadOnlyThreshold"
+	// MessagePoolROThreshold holds descriptive message for PoolROThreshold
+	MessagePoolROThreshold EventReason = "Pool storage limit reached to threshold. Pool expansion is required to make it's replica RW"
 )
 
 // Periodic interval duration.

cmd/cstor-pool-mgmt/controller/pool-controller/handler.go

@@ -35,6 +35,7 @@ import (
 	"github.com/openebs/maya/pkg/util"
 	corev1 "k8s.io/api/core/v1"
 	k8serror "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/tools/cache"
@@ -132,6 +133,7 @@ func (c *CStorPoolController) syncHandler(key string, operation common.QueueOper
 		return err
 	}
 	// Synchronize cstor pool used and free capacity fields on CSP object.
+	// Also verify and handle pool ReadOnly threshold limit
 	// Any kind of sync activity should be done from here.
 	// ToDo: Move status sync (of csp) here from cStorPoolEventHandler function.
 	// ToDo: Instead of having statusSync, capacitySync we can make it generic resource sync which syncs all the
@@ -206,7 +208,10 @@ func (c *CStorPoolController) cStorPoolEventHandler(operation common.QueueOperat
 			return status, err
 		}
 		klog.V(4).Infof("Synchronizing cStor pool status for pool %s", cStorPoolGot.ObjectMeta.Name)
-		status, err := c.getPoolStatus(cStorPoolGot)
+		status, readOnly, err := c.getPoolStatus(cStorPoolGot)
+		if err == nil {
+			cStorPoolGot.Status.ReadOnly = readOnly
+		}
 		return status, err
 	}
 	klog.Errorf("ignored event '%s' for cstor pool '%s'", string(operation), string(cStorPoolGot.ObjectMeta.Name))
@@ -394,14 +399,14 @@ func (c *CStorPoolController) cStorPoolDestroyEventHandler(cStorPoolGot *apis.CS
 }
 
 //  getPoolStatus is a wrapper that fetches the status of cstor pool.
-func (c *CStorPoolController) getPoolStatus(cStorPoolGot *apis.CStorPool) (string, error) {
-	poolStatus, err := pool.Status(string(pool.PoolPrefix) + string(cStorPoolGot.ObjectMeta.UID))
+func (c *CStorPoolController) getPoolStatus(cStorPoolGot *apis.CStorPool) (string, bool, error) {
+	poolStatus, readOnly, err := pool.Status(string(pool.PoolPrefix) + string(cStorPoolGot.ObjectMeta.UID))
 	if err != nil {
 		// ToDO : Put error in event recorder
 		c.recorder.Event(cStorPoolGot, corev1.EventTypeWarning, string(common.FailureStatusSync), string(common.MessageResourceFailStatusSync))
-		return "", err
+		return "", false, err
 	}
-	return poolStatus, nil
+	return poolStatus, readOnly, nil
 }
 
 // getPoolResource returns object corresponding to the resource key
@@ -562,7 +567,74 @@ func (c *CStorPoolController) syncCsp(cStorPool *apis.CStorPool) {
 		c.recorder.Event(cStorPool, corev1.EventTypeWarning, string(common.FailureCapacitySync), string(common.MessageResourceFailCapacitySync))
 	} else {
 		cStorPool.Status.Capacity = *capacity
+		c.updateROMode(cStorPool)
+	}
+}
+
+// updateROMode update pool readOnly mode and csp status
+func (c *CStorPoolController) updateROMode(cStorPool *apis.CStorPool) {
+	// Note: cStorPool status has been updated by handler prior to this call.
+	// So it can be different than etcd version. So below checks are done on latest values.
+	capacity := cStorPool.Status.Capacity
+	rOThresholdLimit := cStorPool.Spec.PoolSpec.ROThresholdLimit
+
+	qn, err := convertToBytes([]string{capacity.Total, capacity.Free, capacity.Used})
+	if err != nil {
+		klog.Errorf("Failed to parse capacity.. err=%s", err)
+		return
+	}
+
+	total, _, used := qn[0], qn[1], qn[2]
+	usedCapacity := (used * 100) / total
+
+	if (int(usedCapacity) >= rOThresholdLimit) &&
+		(rOThresholdLimit != 0 &&
+			rOThresholdLimit != 100) {
+		if !cStorPool.Status.ReadOnly {
+			if err = pool.SetPoolRDMode(cStorPool, true); err != nil {
+				klog.Errorf("Failed to set pool readOnly mode : %v", err)
+			} else {
+				cStorPool.Status.ReadOnly = true
+				c.recorder.Event(cStorPool,
+					corev1.EventTypeWarning,
+					string(common.PoolROThreshold),
+					string(common.MessagePoolROThreshold),
+				)
+			}
+		}
+	} else {
+		if cStorPool.Status.ReadOnly {
+			if err = pool.SetPoolRDMode(cStorPool, false); err != nil {
+				klog.Errorf("Failed to unset pool readOnly mode : %v", err)
+			} else {
+				cStorPool.Status.ReadOnly = false
+			}
+		}
+	}
+	return
+}
+
+func convertToBytes(a []string) (number []int64, err error) {
+	if len(a) == 0 {
+		err = errors.New("empty input")
+		return
+	}
+
+	defer func() {
+		if r := recover(); r != nil {
+			err = errors.New("unable to parse")
+		}
+	}()
+
+	parser := func(s string) int64 {
+		d := resource.MustParse(s + "i")
+		return d.Value()
+	}
+
+	for _, v := range a {
+		number = append(number, parser(v))
 	}
+	return
 }
 
 func (c *CStorPoolController) getDeviceIDs(csp *apis.CStorPool) ([]string, error) {

cmd/cstor-pool-mgmt/pool/pool.go

@@ -390,62 +390,67 @@ func Capacity(poolName string) (*apis.CStorPoolCapacityAttr, error) {
 }
 
 // PoolStatus finds the status of the pool.
-// The ouptut of command(`zpool status <pool-name>`) executed is as follows:
+// The ouptut of command(`zpool get -Hp  -ovalue health,io.openebs:readonly <pool-name>`) executed is as follows:
 
 /*
-		  pool: cstor-530c9c4f-e0df-11e8-94a8-42010a80013b
-	 state: ONLINE
-	  scan: none requested
-	config:
-
-		NAME                                        STATE     READ WRITE CKSUM
-		cstor-530c9c4f-e0df-11e8-94a8-42010a80013b  ONLINE       0     0     0
-		  scsi-0Google_PersistentDisk_ashu-disk2    ONLINE       0     0     0
-
-	errors: No known data errors
+root@cstor-pool-1dvj-854db8dc56-prblp:/# zpool get -Hp  -ovalue health,io.openebs:readonly  cstor-3cbec7b9-578d-11ea-b66e-42010a9a0080
+ONLINE
+off
 */
 // The output is then parsed by poolStatusOutputParser function to get the status of the pool
-func Status(poolName string) (string, error) {
+func Status(poolName string) (string, bool, error) {
 	var poolStatus string
-	statusPoolStr := []string{"status", poolName}
+	var readOnly bool
+
+	statusPoolStr := []string{"get", "-Hp", "-ovalue", "health,io.openebs:readonly", poolName}
 	stdoutStderr, err := RunnerVar.RunCombinedOutput(zpool.PoolOperator, statusPoolStr...)
 	if err != nil {
 		klog.Errorf("Unable to get pool status: %v", string(stdoutStderr))
-		return "", err
-	}
-	poolStatus = poolStatusOutputParser(string(stdoutStderr))
-	if poolStatus == ZpoolStatusDegraded {
-		return string(apis.CStorPoolStatusDegraded), nil
-	} else if poolStatus == ZpoolStatusFaulted {
-		return string(apis.CStorPoolStatusOffline), nil
-	} else if poolStatus == ZpoolStatusOffline {
-		return string(apis.CStorPoolStatusOffline), nil
-	} else if poolStatus == ZpoolStatusOnline {
-		return string(apis.CStorPoolStatusOnline), nil
-	} else if poolStatus == ZpoolStatusRemoved {
-		return string(apis.CStorPoolStatusDegraded), nil
-	} else if poolStatus == ZpoolStatusUnavail {
-		return string(apis.CStorPoolStatusOffline), nil
-	} else {
-		return string(apis.CStorPoolStatusError), nil
-	}
+		return "", readOnly, err
+	}
+	readOnly, poolStatus = poolStatusOutputParser(string(stdoutStderr))
+
+	poolStatus = func(s string) string {
+		switch s {
+		case ZpoolStatusDegraded:
+			return string(apis.CStorPoolStatusDegraded)
+		case ZpoolStatusFaulted:
+			return string(apis.CStorPoolStatusOffline)
+		case ZpoolStatusOffline:
+			return string(apis.CStorPoolStatusOffline)
+		case ZpoolStatusOnline:
+			return string(apis.CStorPoolStatusOnline)
+		case ZpoolStatusRemoved:
+			return string(apis.CStorPoolStatusDegraded)
+		case ZpoolStatusUnavail:
+			return string(apis.CStorPoolStatusError)
+		default:
+			return string(apis.CStorPoolStatusError)
+		}
+	}(poolStatus)
+
+	return poolStatus, readOnly, nil
 }
 
 // poolStatusOutputParser parse output of `zpool status` command to extract the status of the pool.
 // ToDo: Need to find some better way e.g contract for zpool command outputs.
-func poolStatusOutputParser(output string) string {
+func poolStatusOutputParser(output string) (bool, string) {
 	var outputStr []string
 	var poolStatus string
-	if strings.TrimSpace(string(output)) != "" {
-		outputStr = strings.Split(string(output), "\n")
-		if !(len(outputStr) < 2) {
-			poolStatusArr := strings.Split(outputStr[1], ":")
-			if !(len(outputStr) < 2) {
-				poolStatus = strings.TrimSpace(poolStatusArr[1])
-			}
-		}
+	var readOnly bool
+
+	outputStr = strings.Split(string(output), "\n")
+
+	if len(outputStr) != 3 {
+		klog.Errorf("Invalid input='%s' for poolStatusOutputParser", output)
+		return readOnly, poolStatus
+	}
+
+	poolStatus = strings.TrimSpace(string(outputStr[0]))
+	if outputStr[1] == "on" {
+		readOnly = true
 	}
-	return poolStatus
+	return readOnly, poolStatus
 }
 
 // capacityOutputParser parse output of `zpool get` command to extract the capacity of the pool.
@@ -552,3 +557,22 @@ func GetDeviceIDs(csp *apis.CStorPool) ([]string, error) {
 	}
 	return bdDeviceID, nil
 }
+
+// SetPoolRDMode set/unset pool readonly
+func SetPoolRDMode(csp *apis.CStorPool, isROMode bool) error {
+	mode := "off"
+	if isROMode {
+		mode = "on"
+	}
+
+	cmd := []string{"set",
+		"io.openebs:readonly=" + mode,
+		string(PoolPrefix) + string(csp.ObjectMeta.UID),
+	}
+
+	stdoutStderr, err := RunnerVar.RunCombinedOutput(zpool.PoolOperator, cmd...)
+	if err != nil {
+		return errors.Errorf("Failed to update readOnly mode out:%v err:%v", string(stdoutStderr), err)
+	}
+	return nil
+}