Merge pull request #786 from nbalacha/rebalance-plugin

GD2/rebalance: Rebalance plugin

glusterd2/plugin/plugins.go

@@ -9,6 +9,7 @@ import (
 	"github.com/gluster/glusterd2/plugins/georeplication"
 	"github.com/gluster/glusterd2/plugins/glustershd"
 	"github.com/gluster/glusterd2/plugins/quota"
+	"github.com/gluster/glusterd2/plugins/rebalance"
 
 	// ensure init() of non-plugins also gets executed
 	_ "github.com/gluster/glusterd2/plugins/afr"
@@ -23,4 +24,5 @@ var PluginsList = []GlusterdPlugin{
 	&events.Plugin{},
 	&glustershd.Plugin{},
 	&device.Plugin{},
+	&rebalance.Plugin{},
 }

glusterd2/servers/sunrpc/handshake_prog.go

@@ -11,6 +11,7 @@ import (
 	"github.com/gluster/glusterd2/glusterd2/store"
 	"github.com/gluster/glusterd2/glusterd2/volume"
 	"github.com/gluster/glusterd2/pkg/sunrpc"
+	"github.com/gluster/glusterd2/plugins/rebalance"
 
 	log "github.com/sirupsen/logrus"
 )
@@ -22,9 +23,13 @@ const (
 
 const (
 	gfHndskGetSpec       = 2 // GF_HNDSK_GETSPEC
+	gfHndskEventNotify   = 5 // GF_HNDSK_EVENT_NOTIFY,
 	gfHndskGetVolumeInfo = 6 // GF_HNDSK_GET_VOLUME_INFO
 
 )
+const (
+	gfEventNotifyDefragStatus = 0
+)
 
 var volfilePrefix = "volfiles/"
 
@@ -42,6 +47,8 @@ func newGfHandshake() *GfHandshake {
 				ProcedureNumber: gfHndskGetSpec}, Name: "ServerGetspec"},
 			{ID: sunrpc.ProcedureID{ProgramNumber: hndskProgNum, ProgramVersion: hndskProgVersion,
 				ProcedureNumber: gfHndskGetVolumeInfo}, Name: "ServerGetVolumeInfo"},
+			{ID: sunrpc.ProcedureID{ProgramNumber: hndskProgNum, ProgramVersion: hndskProgVersion,
+				ProcedureNumber: gfHndskEventNotify}, Name: "ServerEventNotify"},
 		},
 	}
 }
@@ -244,3 +251,57 @@ Out:
 
 	return nil
 }
+
+// GfServerEventNotifyReq is sent by the rebalance process before it terminates
+// and contains the status information in a dict
+type GfServerEventNotifyReq struct {
+	Op   int
+	Dict []byte
+}
+
+//GfServerEventNotifyResp contains the response to the GfServerEventNotifyReq
+type GfServerEventNotifyResp struct {
+	OpRet   int
+	OpErrno int
+	Dict    []byte
+}
+
+//ServerEventNotify processes the status information sent by the rebalance process
+func (p *GfHandshake) ServerEventNotify(args *GfServerEventNotifyReq, reply *GfServerEventNotifyResp) error {
+
+	var (
+		// pre-declared variables are required for goto statements
+		err error
+	)
+
+	switch args.Op {
+	case gfEventNotifyDefragStatus:
+		reqDict, err := dict.Unserialize(args.Dict)
+		if err != nil {
+			log.WithError(err).Error("dict unserialize failed")
+			reply.OpRet = -1
+			reply.OpErrno = int(syscall.EINVAL)
+			goto Out
+		}
+		err = rebalance.HandleEventNotify(reqDict)
+		if err != nil {
+			reply.OpRet = -1
+			reply.OpErrno = int(syscall.EINVAL)
+			goto Out
+		}
+
+	default:
+		log.WithError(err).Error("Unknown op received in event notify")
+		reply.OpRet = -1
+		reply.OpErrno = int(syscall.EINVAL)
+		goto Out
+	}
+
+Out:
+	if err != nil {
+		reply.OpRet = -1
+	}
+
+	return nil
+
+}

plugins/rebalance/api/req_resp.go

@@ -0,0 +1,74 @@
+package rebalance
+
+import (
+	"github.com/pborman/uuid"
+)
+
+// Status represents Rebalance Status
+type Status uint64
+
+const (
+	// NotStarted should be set only for a node in which rebalance process is not started
+	NotStarted Status = iota
+	// Started should be set only for a node that has been just started rebalance process
+	Started
+	// Stopped should be set only for a node that has been just stopped rebalance process
+	Stopped
+	// Complete should be set only for a node that the rebalance process is completed
+	Complete
+	// Failed should be set only for a node that are failed to run rebalance process
+	Failed
+)
+
+// Command represents Rebalance Commands
+type Command uint64
+
+const (
+	// CmdNone indicates an invalid command
+	CmdNone Command = iota
+	// CmdStart starts the rebalance
+	CmdStart
+	// CmdStop stops the rebalance
+	CmdStop
+	// CmdStatus gets the rebalance status
+	CmdStatus
+	// CmdFixLayoutStart starts a rebalance fix-layout operation
+	CmdFixLayoutStart
+	// CmdStartForce starts rebalance with the force option
+	CmdStartForce
+)
+
+// RebalNodeStatus represents the rebalance status on the Node
+type RebalNodeStatus struct {
+	NodeID            uuid.UUID `json:"nodeid"`
+	Status            string    `json:"status"`
+	RebalancedFiles   string    `json:"rebalanced-files"`
+	RebalancedSize    string    `json:"size"`
+	LookedupFiles     string    `json:"lookedup"`
+	SkippedFiles      string    `json:"skipped"`
+	RebalanceFailures string    `json:"failed"`
+	ElapsedTime       string    `json:"run-time"`
+	TimeLeft          string    `json:"time-left"`
+}
+
+// RebalInfo represents the rebalance operation information
+type RebalInfo struct {
+	Volname     string
+	State       Status
+	Cmd         Command
+	RebalanceID uuid.UUID
+	CommitHash  uint64
+	RebalStats  []RebalNodeStatus
+}
+
+// RebalStatus represents the rebalance status response
+type RebalStatus struct {
+	Volname     string            `json:"volume"`
+	RebalanceID uuid.UUID         `json:"rebalance-id"`
+	Nodes       []RebalNodeStatus `json:"nodes-status"`
+}
+
+// StartReq contains the options passed to the Rebalance Start Request
+type StartReq struct {
+	Option string `json:"option,omitempty"`
+}

plugins/rebalance/errors.go

@@ -0,0 +1,14 @@
+package rebalance
+
+import (
+	"errors"
+)
+
+var (
+	// ErrVolNotDistribute : Cannot run rebalance on a non distribute volume
+	ErrVolNotDistribute = errors.New("Not a distribute volume")
+	// ErrRebalanceNotStarted : Rebalance not started on the volume
+	ErrRebalanceNotStarted = errors.New("Rebalance not started")
+	// ErrRebalanceInvalidOption : Invalid option provided to the rebalance start command
+	ErrRebalanceInvalidOption = errors.New("Invalid Rebalance start option")
+)

plugins/rebalance/eventnotify.go

@@ -0,0 +1,79 @@
+package rebalance
+
+import (
+	"context"
+	"errors"
+	"strings"
+
+	"github.com/gluster/glusterd2/glusterd2/gdctx"
+	"github.com/gluster/glusterd2/glusterd2/transaction"
+	"github.com/gluster/glusterd2/glusterd2/volume"
+	rebalanceapi "github.com/gluster/glusterd2/plugins/rebalance/api"
+
+	log "github.com/sirupsen/logrus"
+)
+
+//HandleEventNotify updates the rebalinfo in the store with the status sent by
+// the rebalance process
+func HandleEventNotify(status map[string]string) error {
+	var (
+		ok      bool
+		volname string
+		err     error
+	)
+
+	volname, ok = status["volname"]
+	if !ok {
+		err = errors.New("volname key not found")
+		return err
+	}
+
+	volname = strings.TrimLeft(volname, "rebalance/")
+	log.Debug("In RebalanceHandleEventNotify " + volname)
+
+	var rebalinfo *rebalanceapi.RebalInfo
+	var rebalNodeStatus rebalanceapi.RebalNodeStatus
+	ctx := context.TODO()
+
+	lock, unlock := transaction.CreateLockFuncs(volname)
+	if err := lock(ctx); err != nil {
+
+		log.WithError(err).Error("Locking failed. Unable to update store")
+		return err
+	}
+	defer unlock(ctx)
+
+	vol, err := volume.GetVolume(volname)
+	if err != nil {
+		return err
+	}
+
+	rebalinfo, err = GetRebalanceInfo(volname)
+	if err != nil {
+		log.WithError(err).Error("Failed to get rebalance info from store")
+		return err
+	}
+
+	rebalNodeStatus.NodeID = gdctx.MyUUID
+	rebalNodeStatus.Status = status["status"]
+	rebalNodeStatus.RebalancedFiles = status["files"]
+	rebalNodeStatus.RebalancedSize = status["size"]
+	rebalNodeStatus.LookedupFiles = status["lookups"]
+	rebalNodeStatus.SkippedFiles = status["skipped"]
+	rebalNodeStatus.RebalanceFailures = status["failures"]
+	rebalNodeStatus.ElapsedTime = status["run-time"]
+	rebalNodeStatus.TimeLeft = status["time-left"]
+
+	rebalinfo.RebalStats = append(rebalinfo.RebalStats, rebalNodeStatus)
+	if len(rebalinfo.RebalStats) == len(vol.Nodes()) {
+		rebalinfo.State = rebalanceapi.Complete
+	}
+
+	err = StoreRebalanceInfo(rebalinfo)
+	if err != nil {
+		log.WithError(err).Error("Failed to store rebalance info")
+		return err
+	}
+
+	return nil
+}

plugins/rebalance/init.go

@@ -0,0 +1,54 @@
+package rebalance
+
+import (
+	"github.com/gluster/glusterd2/glusterd2/servers/rest/route"
+	"github.com/gluster/glusterd2/glusterd2/transaction"
+	"github.com/gluster/glusterd2/pkg/utils"
+	rebalanceapi "github.com/gluster/glusterd2/plugins/rebalance/api"
+)
+
+// Plugin is a structure which implements GlusterdPlugin interface
+type Plugin struct {
+}
+
+// Name returns name of plugin
+func (p *Plugin) Name() string {
+	return "rebalance"
+}
+
+// RestRoutes returns list of REST API routes to register with Glusterd
+func (p *Plugin) RestRoutes() route.Routes {
+	return route.Routes{
+		route.Route{
+			Name:        "RebalanceStart",
+			Method:      "POST",
+			Pattern:     "/volumes/{volname}/rebalance/start",
+			Version:     1,
+			RequestType: utils.GetTypeString((*rebalanceapi.StartReq)(nil)),
+			//			ResponseType: utils.GetTypeString((*rebalanceapi.RebalInfo)(nil)),
+			HandlerFunc: rebalanceStartHandler},
+		route.Route{
+			Name:    "RebalanceStop",
+			Method:  "POST",
+			Pattern: "/volumes/{volname}/rebalance/stop",
+			Version: 1,
+			//			ResponseType: utils.GetTypeString((*rebalanceapi.RebalInfo)(nil)),
+			HandlerFunc: rebalanceStopHandler},
+		route.Route{
+			Name:    "RebalanceStatus",
+			Method:  "GET",
+			Pattern: "/volumes/{volname}/rebalance",
+			Version: 1,
+			//			ResponseType: utils.GetTypeString((*rebalanceapi.RebalInfo)(nil)),
+			HandlerFunc: rebalanceStatusHandler},
+	}
+}
+
+// RegisterStepFuncs registers transaction step functions with
+// Glusterd Transaction framework
+func (p *Plugin) RegisterStepFuncs() {
+	transaction.RegisterStepFunc(txnRebalanceStart, "rebalance-start")
+	transaction.RegisterStepFunc(txnRebalanceStop, "rebalance-stop")
+	transaction.RegisterStepFunc(txnRebalanceStatus, "rebalance-status")
+	transaction.RegisterStepFunc(txnRebalanceStoreDetails, "rebalance-store")
+}