gfproxy: Add plugin for daemon management

glusterd2/plugin/plugins.go

@@ -7,6 +7,7 @@ import (
 	"github.com/gluster/glusterd2/plugins/device"
 	"github.com/gluster/glusterd2/plugins/events"
 	"github.com/gluster/glusterd2/plugins/georeplication"
+	"github.com/gluster/glusterd2/plugins/gfproxyd"
 	"github.com/gluster/glusterd2/plugins/glustershd"
 	"github.com/gluster/glusterd2/plugins/quota"
 )
@@ -18,5 +19,6 @@ var PluginsList = []GlusterdPlugin{
 	&quota.Plugin{},
 	&events.Plugin{},
 	&glustershd.Plugin{},
+	&gfproxyd.Plugin{},
 	&device.Plugin{},
 }

glusterd2/volume/struct.go

@@ -44,6 +44,8 @@ const (
 	VkeyScrubFrequency = "bit-rot.scrub-freq"
 	// VkeyScrubThrottle is the key for controls scrubber throttle
 	VkeyScrubThrottle = "bit-rot.scrub-throttle"
+	//VkeyFeaturesGfproxyd is the key which enables or disables gfproxyd
+	VkeyFeaturesGfproxyd = "features.gfproxyd"
 )
 
 var (

glusterd2/volume/volume-utils.go

@@ -45,3 +45,12 @@ func IsQuotaEnabled(v *Volinfo) bool {
 	}
 	return false
 }
+
+// IsGfproxydEnabled returns true if gfproxyd is enabled for a volume and false otherwise
+func IsGfproxydEnabled(v *Volinfo) bool {
+	val, exists := v.Options[VkeyFeaturesGfproxyd]
+	if exists && val == "on" {
+		return true
+	}
+	return false
+}

pkg/errors/error.go

@@ -38,5 +38,7 @@ var (
 	ErrUnknownValue            = errors.New("unknown value specified")
 	ErrGetFailed               = errors.New("failed to get value from the store")
 	ErrUnmarshallFailed        = errors.New("failed to unmarshall from json")
+	ErrGfproxydAlreadyEnabled  = errors.New("Gfproxyd is already enabled")
+	ErrGfproxydAlreadyDisabled = errors.New("Gfproxyd is already disabled")
 	ErrClusterNotFound         = errors.New("Cluster instance not found in store")
 )

plugins/gfproxyd/gfproxyd.go

@@ -0,0 +1,92 @@
+package gfproxyd
+
+import (
+	"bytes"
+	"fmt"
+	"net"
+	"os/exec"
+	"path"
+	"strconv"
+
+	"github.com/gluster/glusterd2/glusterd2/pmap"
+	config "github.com/spf13/viper"
+)
+
+const (
+	gfproxydBin = "glusterfsd"
+)
+
+// gfproxyd type represents information about gfproxyd process
+type gfproxyd struct {
+	volname      string
+	args         string
+	pidfilepath  string
+	binarypath   string
+	volfileID    string
+	logfile      string
+	gfproxyID    string
+	gfproxydPort string
+}
+
+// Name returns human-friendly name of the gfproxyd process. This is used for logging.
+func (g *gfproxyd) Name() string {
+	return g.gfproxyID
+}
+
+// Path returns absolute path to the binary of gfproxyd process
+func (g *gfproxyd) Path() string {
+	return g.binarypath
+}
+
+// Args returns arguments to be passed to gfproxyd process during spawn.
+func (g *gfproxyd) Args() string {
+	return g.args
+}
+
+// SocketFile returns path to the socket file
+func (g *gfproxyd) SocketFile() string {
+	return ""
+}
+
+// PidFile returns path to the pid file of the gfproxyd process
+func (g *gfproxyd) PidFile() string {
+	return g.pidfilepath
+}
+
+// newgfproxyd returns a new instance of gfproxyd type which implements the Daemon interface
+func newgfproxyd(volname string) (*gfproxyd, error) {
+	g := &gfproxyd{volname: volname}
+	binarypath, e := exec.LookPath(gfproxydBin)
+	if e != nil {
+		return nil, e
+	}
+	g.binarypath = binarypath
+	g.gfproxyID = fmt.Sprintf("gfproxyd-%s", volname)
+	g.gfproxydPort = strconv.Itoa(pmap.AssignPort(0, g.gfproxyID))
+	g.volfileID = fmt.Sprintf("gfproxyd/%s", volname)
+	g.logfile = path.Join(config.GetString("logdir"), "glusterfs", fmt.Sprintf("gfproxyd-%s.log", volname))
+	g.pidfilepath = fmt.Sprintf("%s/gfproxyd-%s.pid", config.GetString("rundir"), volname)
+
+	shost, _, _ := net.SplitHostPort(config.GetString("clientaddress"))
+	if shost == "" {
+		shost = "localhost"
+	}
+
+	var buffer bytes.Buffer
+	buffer.WriteString(fmt.Sprintf(" -s %s", shost))
+	buffer.WriteString(fmt.Sprintf(" --volfile-id %s", g.volfileID))
+	buffer.WriteString(fmt.Sprintf(" -p %s", g.pidfilepath))
+	buffer.WriteString(fmt.Sprintf(" -l %s", g.logfile))
+	buffer.WriteString(fmt.Sprintf(" --brick-name %s", g.gfproxyID))
+	buffer.WriteString(fmt.Sprintf(" --brick-port %s", g.gfproxydPort))
+	buffer.WriteString(fmt.Sprintf(" --xlator-option %s-server.listen-port=%s", g.volname, g.gfproxydPort))
+
+	g.args = buffer.String()
+
+	return g, nil
+}
+
+// ID returns the unique identifier of the gfproxyd.
+func (g *gfproxyd) ID() string {
+	return g.gfproxyID
+}

plugins/gfproxyd/init.go

@@ -0,0 +1,46 @@
+package gfproxyd
+
+import (
+	"github.com/gluster/glusterd2/glusterd2/servers/rest/route"
+	"github.com/gluster/glusterd2/glusterd2/transaction"
+	"github.com/gluster/glusterd2/pkg/sunrpc"
+)
+
+// Plugin is a structure which implements GlusterdPlugin interface
+type Plugin struct {
+}
+
+// Name returns name of plugin
+func (p *Plugin) Name() string {
+	return "gfproxyd"
+}
+
+// SunRPCProgram returns sunrpc program to register with Glusterd
+func (p *Plugin) SunRPCProgram() sunrpc.Program {
+	return nil
+}
+
+// RestRoutes returns list of REST API routes to register with Glusterd
+func (p *Plugin) RestRoutes() route.Routes {
+	return route.Routes{
+		route.Route{
+			Name:        "GfproxydEnable",
+			Method:      "POST",
+			Pattern:     "/volumes/{name}/gfproxy/enable",
+			Version:     1,
+			HandlerFunc: gfproxydEnableHandler},
+		route.Route{
+			Name:        "GfproxydDisable",
+			Method:      "POST",
+			Pattern:     "/volumes/{name}/gfproxy/disable",
+			Version:     1,
+			HandlerFunc: gfproxydDisableHandler},
+	}
+}
+
+// RegisterStepFuncs registers transaction step functions with
+// Glusterd Transaction framework
+func (p *Plugin) RegisterStepFuncs() {
+	transaction.RegisterStepFunc(txnGfproxydStart, "gfproxyd-start.Commit")
+	transaction.RegisterStepFunc(txnGfproxydStop, "gfproxyd-stop.Commit")
+}

plugins/gfproxyd/rest.go

@@ -0,0 +1,164 @@
+package gfproxyd
+
+import (
+	"net/http"
+
+	"github.com/gluster/glusterd2/glusterd2/gdctx"
+	restutils "github.com/gluster/glusterd2/glusterd2/servers/rest/utils"
+	"github.com/gluster/glusterd2/glusterd2/transaction"
+	"github.com/gluster/glusterd2/glusterd2/volume"
+	"github.com/gluster/glusterd2/pkg/api"
+	"github.com/gluster/glusterd2/pkg/errors"
+
+	"github.com/gorilla/mux"
+	log "github.com/sirupsen/logrus"
+
+	"github.com/pborman/uuid"
+)
+
+func gfproxydEnableHandler(w http.ResponseWriter, r *http.Request) {
+	// Collect inputs from URL
+	p := mux.Vars(r)
+	volname := p["name"]
+
+	ctx := r.Context()
+	logger := gdctx.GetReqLogger(ctx)
+
+	//validate volume name
+	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 gfproxyd is already enabled
+	if volume.IsGfproxydEnabled(volinfo) {
+		restutils.SendHTTPError(ctx, w, http.StatusBadRequest, errors.ErrGfproxydAlreadyEnabled.Error(), api.ErrCodeDefault)
+		return
+	}
+
+	// Transaction which starts gfproxyd on all nodes.
+	txn := transaction.NewTxn(ctx)
+	defer txn.Cleanup()
+
+	txn.Ctx.Set("volname", volname)
+
+	// Enable gfproxyd
+	volinfo.Options[volume.VkeyFeaturesGfproxyd] = "on"
+
+	if err := txn.Ctx.Set("volinfo", volinfo); err != nil {
+		logger.WithError(err).Error("failed to set volinfo in transaction context")
+		restutils.SendHTTPError(ctx, w, http.StatusInternalServerError, err.Error(), api.ErrCodeDefault)
+		return
+	}
+
+	//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: "vol-option.UpdateVolinfo",
+			Nodes:  []uuid.UUID{gdctx.MyUUID},
+		},
+		{
+			DoFunc: "gfproxyd-start.Commit",
+			Nodes:  txn.Nodes,
+		},
+		unlock,
+	}
+
+	err = txn.Do()
+	if err != nil {
+		logger.WithFields(log.Fields{
+			"error":   err.Error(),
+			"volname": volname,
+		}).Error("failed to start gfproxy daemon")
+		restutils.SendHTTPError(ctx, w, http.StatusInternalServerError, err.Error(), api.ErrCodeDefault)
+		return
+	}
+
+	restutils.SendHTTPResponse(ctx, w, http.StatusOK, nil)
+}
+
+func gfproxydDisableHandler(w http.ResponseWriter, r *http.Request) {
+	// Collect inputs from URL
+	p := mux.Vars(r)
+	volname := p["name"]
+
+	ctx := r.Context()
+	logger := gdctx.GetReqLogger(ctx)
+
+	//validate volume name
+	volinfo, err := volume.GetVolume(volname)
+	if err != nil {
+		restutils.SendHTTPError(ctx, w, http.StatusNotFound, errors.ErrVolNotFound.Error(), api.ErrCodeDefault)
+		return
+	}
+
+	// Check if gfproxyd is already disabled
+	if !volume.IsGfproxydEnabled(volinfo) {
+		restutils.SendHTTPError(ctx, w, http.StatusBadRequest, errors.ErrGfproxydAlreadyDisabled.Error(), api.ErrCodeDefault)
+		return
+	}
+
+	// Transaction which stop gfproxyd on all nodes.
+	txn := transaction.NewTxn(ctx)
+	defer txn.Cleanup()
+
+	txn.Ctx.Set("volname", volname)
+
+	// Disable gfproxyd
+	volinfo.Options[volume.VkeyFeaturesGfproxyd] = "off"
+
+	if err := txn.Ctx.Set("volinfo", volinfo); err != nil {
+		logger.WithError(err).Error("failed to set volinfo in transaction context")
+		restutils.SendHTTPError(ctx, w, http.StatusInternalServerError, err.Error(), api.ErrCodeDefault)
+		return
+	}
+
+	//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: "vol-option.UpdateVolinfo",
+			Nodes:  []uuid.UUID{gdctx.MyUUID},
+		},
+		{
+			DoFunc: "gfproxyd-stop.Commit",
+			Nodes:  txn.Nodes,
+		},
+		unlock,
+	}
+
+	err = txn.Do()
+	if err != nil {
+		logger.WithFields(log.Fields{
+			"error":   err.Error(),
+			"volname": volname,
+		}).Error("failed to start gfproxy daemon")
+		restutils.SendHTTPError(ctx, w, http.StatusInternalServerError, err.Error(), api.ErrCodeDefault)
+		return
+	}
+
+	restutils.SendHTTPResponse(ctx, w, http.StatusOK, nil)
+
+}

plugins/gfproxyd/transactions.go

@@ -0,0 +1,38 @@
+package gfproxyd
+
+import (
+	"github.com/gluster/glusterd2/glusterd2/daemon"
+	"github.com/gluster/glusterd2/glusterd2/transaction"
+)
+
+func txnGfproxydStart(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
+	}
+
+	gfproxyd, err := newgfproxyd(volname)
+	if err != nil {
+		return err
+	}
+	err = daemon.Start(gfproxyd, true)
+	return err
+}
+
+func txnGfproxydStop(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
+	}
+
+	gfproxyd, err := newgfproxyd(volname)
+	if err != nil {
+		return err
+	}
+	err = daemon.Stop(gfproxyd, true)
+	return err
+}