Merge pull request #536 from kotreshhr/bitrot-scrub-ondemand

bitrot: Add scrub ondemand support

glusterd2/servers/sunrpc/dict.go → glusterd2/servers/sunrpc/dict/dict.go

@@ -1,4 +1,4 @@
-package sunrpc
+package dict
 
 import (
 	"bytes"
@@ -22,10 +22,10 @@ const (
 	dictHeaderLen = 4
 )
 
-// DictUnserialize unmarshals a slice of bytes into a map[string]string
+// Unserialize unmarshals a slice of bytes into a map[string]string
 // Consumers of the map should typecast/extract information from the
 // map values which are of string type
-func DictUnserialize(buf []byte) (map[string]string, error) {
+func Unserialize(buf []byte) (map[string]string, error) {
 
 	newDict := make(map[string]string)
 	tmpHeader := make([]byte, dictHeaderLen)
@@ -67,8 +67,8 @@ func DictUnserialize(buf []byte) (map[string]string, error) {
 	return newDict, nil
 }
 
-// DictSerialize marshals a map[string]string into a slice of bytes.
-func DictSerialize(dict map[string]string) ([]byte, error) {
+// Serialize marshals a map[string]string into a slice of bytes.
+func Serialize(dict map[string]string) ([]byte, error) {
 
 	dictSerializedSize, err := getSerializedDictLen(dict)
 	if err != nil {

glusterd2/servers/sunrpc/handshake_prog.go

@@ -7,6 +7,7 @@ import (
 	"strings"
 	"syscall"
 
+	"github.com/gluster/glusterd2/glusterd2/servers/sunrpc/dict"
 	"github.com/gluster/glusterd2/glusterd2/store"
 	"github.com/gluster/glusterd2/glusterd2/volume"
 
@@ -88,9 +89,9 @@ func (p *GfHandshake) ServerGetspec(args *GfGetspecReq, reply *GfGetspecRsp) err
 	var err error
 	var fileContents []byte
 
-	_, err = DictUnserialize(args.Xdata)
+	_, err = dict.Unserialize(args.Xdata)
 	if err != nil {
-		log.WithError(err).Error("ServerGetspec(): DictUnserialize() failed")
+		log.WithError(err).Error("ServerGetspec(): dict.Unserialize() failed")
 	}
 
 	// Get Volfile from store
@@ -156,9 +157,9 @@ func (p *GfHandshake) ServerGetVolumeInfo(args *GfGetVolumeInfoReq, reply *GfGet
 	)
 	respDict := make(map[string]string)
 
-	reqDict, err := DictUnserialize(args.Dict)
+	reqDict, err := dict.Unserialize(args.Dict)
 	if err != nil {
-		log.WithError(err).Error("DictUnserialize() failed")
+		log.WithError(err).Error("dict unserialize failed")
 		goto Out
 	}
 
@@ -191,7 +192,7 @@ func (p *GfHandshake) ServerGetVolumeInfo(args *GfGetVolumeInfoReq, reply *GfGet
 		respDict["volume_id"] = volinfo.ID.String()
 	}
 
-	reply.Dict, err = DictSerialize(respDict)
+	reply.Dict, err = dict.Serialize(respDict)
 	if err != nil {
 		log.WithError(err).Error("failed to serialize dict")
 	}

glusterd2/volgen2/volfile_scrub.go

@@ -17,7 +17,7 @@ func generateScrubVolfile(volfile *Volfile, clusterinfo []*volume.Volinfo, nodei
 		//If bitrot not enabled for volume, then skip those bricks
 		val, exists := vol.Options[volume.VkeyFeaturesBitrot]
 		if exists && val == "on" {
-			name := fmt.Sprintf("%s-scrub-%d", vol.Name, volIdx)
+			name := fmt.Sprintf("%s-bit-rot-%d", vol.Name, volIdx)
 			scrubvol := scrub.Add("features/bit-rot", vol, nil).SetName(name)
 			clusterGraph(scrubvol, vol, nodeid, &clusterGraphFilters{onlyLocalBricks: true})
 		}

glusterd2/volume/volume-utils.go

@@ -27,3 +27,12 @@ func isBrickPathAvailable(nodeID uuid.UUID, brickPath string) error {
 	}
 	return nil
 }
+
+// IsBitrotEnabled returns true if bitrot is enabled for a volume and false otherwise
+func IsBitrotEnabled(v *Volinfo) bool {
+	val, exists := v.Options[VkeyFeaturesBitrot]
+	if exists && val == "on" {
+		return true
+	}
+	return false
+}

pkg/errors/error.go

@@ -34,6 +34,7 @@ var (
 	ErrProcessAlreadyRunning   = errors.New("Process is already running")
 	ErrBitrotAlreadyEnabled    = errors.New("Bitrot is already enabled")
 	ErrBitrotAlreadyDisabled   = errors.New("Bitrot is already disabled")
+	ErrBitrotNotEnabled        = errors.New("Bitrot is not enabled")
 	ErrUnknownValue            = errors.New("unknown value specified")
 	ErrGetFailed               = errors.New("failed to get value from the store")
 	ErrUnmarshallFailed        = errors.New("failed to unmarshall from json")

plugins/bitrot/init.go

@@ -35,6 +35,12 @@ func (p *Plugin) RestRoutes() route.Routes {
 			Pattern:     "/volumes/{volname}/bitrot/disable",
 			Version:     1,
 			HandlerFunc: bitrotDisableHandler},
+		route.Route{
+			Name:        "ScrubOndemand",
+			Method:      "POST",
+			Pattern:     "/volumes/{volname}/bitrot/scrubondemand",
+			Version:     1,
+			HandlerFunc: bitrotScrubOndemandHandler},
 	}
 }
 
@@ -43,5 +49,6 @@ func (p *Plugin) RestRoutes() route.Routes {
 func (p *Plugin) RegisterStepFuncs() {
 	transaction.RegisterStepFunc(txnBitrotEnableDisable, "bitrot-enable.Commit")
 	transaction.RegisterStepFunc(txnBitrotEnableDisable, "bitrot-disable.Commit")
+	transaction.RegisterStepFunc(txnBitrotScrubOndemand, "bitrot-scrubondemand.Commit")
 	return
 }

plugins/bitrot/rest.go

@@ -37,8 +37,7 @@ func bitrotEnableHandler(w http.ResponseWriter, r *http.Request) {
 	}
 
 	// Check if bitrot is already enabled
-	val, exists := volinfo.Options[volume.VkeyFeaturesBitrot]
-	if exists && val == "on" {
+	if volume.IsBitrotEnabled(volinfo) {
 		restutils.SendHTTPError(ctx, w, http.StatusBadRequest, errors.ErrBitrotAlreadyEnabled.Error(), api.ErrCodeDefault)
 		return
 	}
@@ -100,7 +99,7 @@ func bitrotEnableHandler(w http.ResponseWriter, r *http.Request) {
 		restutils.SendHTTPError(ctx, w, http.StatusInternalServerError, err.Error(), api.ErrCodeDefault)
 		return
 	}
-	restutils.SendHTTPResponse(ctx, w, http.StatusOK, "bitrot enabled")
+	restutils.SendHTTPResponse(ctx, w, http.StatusOK, "Bitrot enabled successfully")
 }
 
 func bitrotDisableHandler(w http.ResponseWriter, r *http.Request) {
@@ -119,8 +118,7 @@ func bitrotDisableHandler(w http.ResponseWriter, r *http.Request) {
 	}
 
 	// Check if bitrot is already disabled
-	val, exists := volinfo.Options[volume.VkeyFeaturesBitrot]
-	if !exists || val == "off" {
+	if !volume.IsBitrotEnabled(volinfo) {
 		restutils.SendHTTPError(ctx, w, http.StatusBadRequest, errors.ErrBitrotAlreadyDisabled.Error(), api.ErrCodeDefault)
 		return
 	}
@@ -177,5 +175,66 @@ func bitrotDisableHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	restutils.SendHTTPResponse(ctx, w, http.StatusOK, "bitrot Disable")
+	restutils.SendHTTPResponse(ctx, w, http.StatusOK, "Bitrot disabled successfully")
+}
+
+func bitrotScrubOndemandHandler(w http.ResponseWriter, r *http.Request) {
+	// Collect inputs from URL
+	p := mux.Vars(r)
+	volName := p["volname"]
+
+	ctx := r.Context()
+	logger := gdctx.GetReqLogger(ctx)
+
+	// Validate volume existence
+	volinfo, err := volume.GetVolume(volName)
+	if err != nil {
+		restutils.SendHTTPError(ctx, w, http.StatusNotFound, errors.ErrVolNotFound.Error(), api.ErrCodeDefault)
+		return
+	}
+
+	// Check if volume is started
+	if volinfo.State != volume.VolStarted {
+		restutils.SendHTTPError(ctx, w, http.StatusBadRequest, errors.ErrVolNotStarted.Error(), api.ErrCodeDefault)
+		return
+	}
+
+	// Check if bitrot is disabled
+	if !volume.IsBitrotEnabled(volinfo) {
+		restutils.SendHTTPError(ctx, w, http.StatusBadRequest, errors.ErrBitrotNotEnabled.Error(), api.ErrCodeDefault)
+		return
+	}
+
+	// Transaction which starts bitd and scrubber on all nodes.
+	txn := transaction.NewTxn(ctx)
+	defer txn.Cleanup()
+
+	//Lock on Volume Name
+	lock, unlock, err := transaction.CreateLockSteps(volName)
+	if err != nil {
+		restutils.SendHTTPError(ctx, w, http.StatusInternalServerError, err.Error(), api.ErrCodeDefault)
+		return
+	}
+
+	txn.Nodes = volinfo.Nodes()
+	txn.Steps = []*transaction.Step{
+		lock,
+		{
+			DoFunc: "bitrot-scrubondemand.Commit",
+			Nodes:  txn.Nodes,
+		},
+		unlock,
+	}
+	txn.Ctx.Set("volname", volName)
+
+	err = txn.Do()
+	if err != nil {
+		logger.WithFields(log.Fields{
+			"error":   err.Error(),
+			"volname": volName,
+		}).Error("failed to start scrubber")
+		restutils.SendHTTPError(ctx, w, http.StatusInternalServerError, err.Error(), api.ErrCodeDefault)
+		return
+	}
+	restutils.SendHTTPResponse(ctx, w, http.StatusOK, "Scrubber started successfully")
 }

plugins/bitrot/transactions.go

@@ -1,7 +1,9 @@
 package bitrot
 
 import (
+	"github.com/gluster/glusterd2/glusterd2/brick"
 	"github.com/gluster/glusterd2/glusterd2/daemon"
+	"github.com/gluster/glusterd2/glusterd2/servers/sunrpc/dict"
 	"github.com/gluster/glusterd2/glusterd2/transaction"
 	"github.com/gluster/glusterd2/glusterd2/volume"
 	log "github.com/sirupsen/logrus"
@@ -97,3 +99,48 @@ error:
 	//TODO: Handle failure of scrubd. bitd should be stopped. Should it be handled in txn undo func
 	return err
 }
+
+func txnBitrotScrubOndemand(c transaction.TxnCtx) error {
+
+	var volname string
+	if err := c.Get("volname", &volname); err != nil {
+		c.Logger().WithError(err).WithField(
+			"key", "volname").Error("failed to get value for key from context")
+		return err
+	}
+
+	scrubDaemon, err := newScrubd()
+	if err != nil {
+		return err
+	}
+
+	c.Logger().WithFields(log.Fields{"volume": volname}).Info("Starting scrubber")
+
+	client, err := daemon.GetRPCClient(scrubDaemon)
+	if err != nil {
+		c.Logger().WithError(err).WithField(
+			"volume", volname).Error("failed to connect to scrubd")
+		return err
+	}
+
+	reqDict := make(map[string]string)
+	reqDict["scrub-value"] = "ondemand"
+	req := &brick.GfBrickOpReq{
+		Name: volname,
+		Op:   int(brick.OpNodeBitrot),
+	}
+	req.Input, err = dict.Serialize(reqDict)
+	if err != nil {
+		c.Logger().WithError(err).WithField(
+			"volume", volname).Error("failed to serialize dict for scrub-value")
+	}
+	var rsp brick.GfBrickOpRsp
+	err = client.Call("Brick.OpNodeBitrot", req, &rsp)
+	if err != nil || rsp.OpRet != 0 {
+		c.Logger().WithError(err).WithField(
+			"volume", volname).Error("failed to send scrubondemand RPC")
+		return err
+	}
+
+	return nil
+}