WIP: Intelligent Volume provisioning

glusterd2/commands/volumes/volume-create.go

@@ -17,6 +17,20 @@ import (
 	"github.com/pborman/uuid"
 )
 
+
+// VolCreateRequest defines the parameters for creating a volume in the volume-create command
+type VolCreateRequest struct {
+	Name         string            `json:"name"`
+	Transport    string            `json:"transport,omitempty"`
+	ReplicaCount int               `json:"replica,omitempty"`
+	Bricks       []string          `json:"bricks,omitempty"`
+	Force        bool              `json:"force,omitempty"`
+	Options      map[string]string `json:"options,omitempty"`
+	Size         int               `json:"size,omitempty"`
+	// Bricks list is ordered (like in glusterd1) and decides which bricks
+	// form replica sets.
+}
+
 func unmarshalVolCreateRequest(msg *api.VolCreateReq, r *http.Request) (int, error) {
 	if err := restutils.UnmarshalRequest(r, msg); err != nil {
 		return 422, gderrors.ErrJSONParsingFailed

glusterd2/middleware/heketi.go

@@ -0,0 +1,53 @@
+package middleware
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"strings"
+
+	"github.com/gluster/glusterd2/glusterd2/commands/volumes"
+	"github.com/gluster/glusterd2/pkg/utils"
+)
+
+// TODO
+// In Go, the idiomatic and recommended way to attach any request scoped
+// metadata information across goroutine and process boundaries is to use the
+// 'context' package. This is not useful unless we pass down this context
+// all through-out the request scope across nodes, and this involves some
+// code changes in function signatures at many places
+// The following simple implementation is good enough until then...
+
+// Heketi is a middleware which generates adds bricks to a volume
+// request if it has a key asking for auto brick allocation. It modifies the
+// HTTP request and adds bricks to it.
+func Heketi(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+
+
+		if r.URL.Path == "/v1/volumes" && r.Method == http.MethodPost {
+			req := new(volumecommands.VolCreateRequest)
+			utils.GetJSONFromRequest(r, req)
+
+			//if (len(req.Bricks) <= 0) && (req.Size > 0) {
+			if req.Size > 0 {
+				replacer := strings.NewReplacer("export", "testexport")
+                                for i :=0; i < len(req.Bricks); i++ {
+				    req.Bricks[i] = replacer.Replace(req.Bricks[i])
+                                }
+				newbody, err := json.Marshal(req)
+				if err != nil {
+					fmt.Printf("Marshalling Error %v", err)
+				}
+
+				r.Body = ioutil.NopCloser(bytes.NewReader(newbody))
+				r.ContentLength = int64(len(newbody))
+
+			}
+
+		}
+		next.ServeHTTP(w, r)
+	})
+}

glusterd2/plugin/plugins.go

@@ -4,11 +4,13 @@ package plugin
 
 import (
 	"github.com/gluster/glusterd2/plugins/georeplication"
+        "github.com/gluster/glusterd2/plugins/heketi"
 	"github.com/gluster/glusterd2/plugins/hello"
 )
 
 // PluginsList is a list of plugins which implements GlusterdPlugin interface
 var PluginsList = []GlusterdPlugin{
 	&hello.Plugin{},
 	&georeplication.Plugin{},
+        &heketi.Plugin{},
 }

glusterd2/servers/rest/rest.go

@@ -84,7 +84,7 @@ func NewMuxed(m cmux.CMux) *GDRest {
 		middleware.ReqIDGenerator,
 		middleware.LogRequest,
 		middleware.Auth,
-	).Then(rest.Routes)
+	        middleware.Heketi).Then(rest.Routes)
 
 	return rest
 }

glusterd2/servers/rest/utils/utils.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/gluster/glusterd2/glusterd2/gdctx"
 	"github.com/gluster/glusterd2/pkg/api"
+        log "github.com/sirupsen/logrus"
 )
 
 // APIError is the placeholder for error string to report back to the client
@@ -59,3 +60,11 @@ func SendHTTPError(ctx context.Context, w http.ResponseWriter, statusCode int, e
 		logger.WithError(err).Error("Failed to send the response -", resp)
 	}
 }
+
+
+// GetReqIDandLogger returns a request ID and a request-scoped logger having
+// the request ID as a logging field.
+func GetReqIDandLogger(r *http.Request) (string, *log.Entry) {
+	reqID := r.Header.Get("X-Request-ID")
+	return reqID, log.WithField("reqid", reqID)
+}

glusterd2/store/store.go

@@ -17,6 +17,7 @@ import (
 )
 
 const (
+        GlusterPrefix = "gluster/"
 	sessionTTL = 30 // used for etcd mutexes and liveness key
 )
 

pkg/utils/jsonutils.go

@@ -0,0 +1,35 @@
+package utils
+
+import (
+	"encoding/json"
+	"io"
+	"io/ioutil"
+	"net/http"
+)
+
+func jsonFromBody(r io.Reader, v interface{}) error {
+
+	// Check body
+	body, err := ioutil.ReadAll(r)
+	if err != nil {
+		return err
+	}
+	if err := json.Unmarshal(body, v); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// GetJSONFromRequest unmarshals JSON in `r` into `v`
+func GetJSONFromRequest(r *http.Request, v interface{}) error {
+	defer r.Body.Close()
+	return jsonFromBody(r.Body, v)
+}
+
+// GetJSONFromResponse unmarshals JSON in `r` into `v`
+func GetJSONFromResponse(r *http.Response, v interface{}) error {
+	defer r.Body.Close()
+	return jsonFromBody(r.Body, v)
+}
+

plugins/heketi/api/req.go

@@ -0,0 +1,7 @@
+package heketi
+
+// AddDeviceReq represents REST API request to add device to heketi managed device list
+type AddDeviceReq struct {
+	NodeID     string `json:"nodeid"`
+	DeviceName string `json:"devicename"`
+}

plugins/heketi/api/resp.go

@@ -0,0 +1,21 @@
+package heketi
+
+import "github.com/pborman/uuid"
+
+const (
+	// HeketiDeviceEnabled represents enabled
+	HeketiDeviceEnabled = "Enabled"
+
+	// HeketiDeviceFrozen represents frozen
+	HeketiDeviceFrozen = "Frozen"
+
+	// HeketiDeviceEvacuated represents evacuated
+	HeketiDeviceEvacuated = "Evacuated"
+)
+
+// DeviceInfo is the added device info
+type DeviceInfo struct {
+	NodeID     uuid.UUID `json:"nodeid"`
+	DeviceName string    `json:"devicename"`
+	State      string    `json:"devicestate"`
+}

plugins/heketi/errors.go

@@ -0,0 +1,8 @@
+package heketi
+
+// ErrHeketiSingleError only error for now
+type ErrHeketiSingleError struct{}
+
+func (e *ErrHeketiSingleError) Error() string {
+	return "Single error"
+}

plugins/heketi/init.go

@@ -0,0 +1,40 @@
+package heketi
+
+import (
+	"github.com/gluster/glusterd2/glusterd2/servers/rest/route"
+	"github.com/gluster/glusterd2/glusterd2/transaction"
+	"github.com/prashanthpai/sunrpc"
+)
+
+// Plugin is a structure which implements GlusterdPlugin interface
+type Plugin struct {
+}
+
+// Name returns name of plugin
+func (p *Plugin) Name() string {
+	return "heketi"
+}
+
+// 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:        "HeketiDeviceAdd",
+			Method:      "POST",
+			Pattern:     "/heketi/{nodeid}/{devicename}/add",
+			Version:     1,
+			HandlerFunc: heketiDeviceAddHandler},
+	}
+}
+
+// RegisterStepFuncs registers transaction step functions with
+// Glusterd Transaction framework
+func (p *Plugin) RegisterStepFuncs() {
+	transaction.RegisterStepFunc(txnHeketiPrepareDevice, "heketi-prepare-device.Commit")
+	transaction.RegisterStepFunc(txnHeketiCreateBrick, "heketi-create-brick.Commit")
+}

plugins/heketi/rest.go

@@ -0,0 +1,100 @@
+package heketi
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+
+	"github.com/pborman/uuid"
+
+	heketiapi "github.com/gluster/glusterd2/plugins/heketi/api"
+	restutils "github.com/gluster/glusterd2/glusterd2/servers/rest/utils"
+	"github.com/gluster/glusterd2/glusterd2/store"
+        "github.com/gluster/glusterd2/pkg/api"
+	"github.com/gluster/glusterd2/glusterd2/transaction"
+
+	"github.com/gorilla/mux"
+	log "github.com/sirupsen/logrus"
+)
+
+const (
+	heketiPrefix string = store.GlusterPrefix + "heketi/"
+)
+
+func heketiDeviceAddHandler(w http.ResponseWriter, r *http.Request) {
+	// Collect inputs from URL
+	p := mux.Vars(r)
+	nodeIDRaw := p["nodeid"]
+	deviceName := p["devicename"]
+	var deviceinfo heketiapi.DeviceInfo
+	deviceinfo.DeviceName = deviceName
+        ctx := r.Context()
+	nodeID := uuid.Parse(nodeIDRaw)
+	if nodeID == nil {
+		restutils.SendHTTPError(ctx, w, http.StatusBadRequest, "Invalid Node ID", api.ErrCodeDefault)
+		return
+	}
+	deviceinfo.NodeID = nodeID
+
+	reqID, logger := restutils.GetReqIDandLogger(r)
+
+	_, err := store.Store.Get(context.TODO(), deviceinfo.NodeID.String())
+	if err != nil {
+		restutils.SendHTTPError(ctx,w, http.StatusInternalServerError, err.Error(), api.ErrCodeDefault)
+		return
+	}
+
+	txn := transaction.NewTxn(ctx)
+	defer txn.Cleanup()
+
+	lock, unlock, err := transaction.CreateLockSteps(deviceinfo.NodeID.String())
+	if err != nil {
+		restutils.SendHTTPError(ctx, w, http.StatusInternalServerError, err.Error(), api.ErrCodeDefault)
+		return
+	}
+
+	nodes := make([]uuid.UUID, 0)
+	nodes = append(nodes, deviceinfo.NodeID)
+
+	txn.Nodes = nodes
+	txn.Steps = []*transaction.Step{
+		lock,
+		{
+			DoFunc: "heketi-prepare-device.Commit",
+			Nodes:  txn.Nodes,
+		},
+		unlock,
+	}
+	txn.Ctx.Set("nodeid", deviceinfo.NodeID.String())
+	txn.Ctx.Set("devicename", deviceinfo.DeviceName)
+
+	err = txn.Do()
+	if err != nil {
+		logger.WithFields(log.Fields{
+			"error":      err.Error(),
+			"nodeid":     deviceinfo.NodeID,
+			"devicename": deviceinfo.DeviceName,
+		}).Error("Failed to prepare device")
+		restutils.SendHTTPError(ctx, w, http.StatusInternalServerError, err.Error(), api.ErrCodeDefault)
+		return
+	}
+
+	// update device state
+	deviceinfo.State = heketiapi.HeketiDeviceEnabled
+
+	json, err := json.Marshal(deviceinfo)
+	if err != nil {
+		log.WithField("error", err).Error("Failed to marshal the DeviceInfo object" + reqID)
+		restutils.SendHTTPError(ctx, w, http.StatusInternalServerError, err.Error(), api.ErrCodeDefault)
+		return
+	}
+
+	_, err = store.Store.Put(context.TODO(), heketiPrefix+"/"+deviceinfo.NodeID.String()+"/"+deviceName, string(json))
+	if err != nil {
+		log.WithError(err).Error("Couldn't add deviceinfo to store")
+		restutils.SendHTTPError(ctx, w, http.StatusInternalServerError, err.Error(), api.ErrCodeDefault)
+		return
+	}
+
+	restutils.SendHTTPResponse(ctx, w, http.StatusOK, deviceinfo)
+}

plugins/heketi/transactions.go

@@ -0,0 +1,37 @@
+package heketi
+
+import (
+	"fmt"
+	"os/exec"
+
+	heketi "github.com/gluster/glusterd2/plugins/heketi/api"
+	"github.com/gluster/glusterd2/glusterd2/transaction"
+)
+
+func txnHeketiPrepareDevice(c transaction.TxnCtx) error {
+	var deviceinfo heketi.DeviceInfo
+
+	if err := c.Get("nodeid", &deviceinfo.NodeID); err != nil {
+		return err
+	}
+	if err := c.Get("devicename", &deviceinfo.DeviceName); err != nil {
+		return err
+	}
+
+
+	cmd1 := exec.Command("pvcreate", "--metadatasize=128M", "--dataalignment=256K", "/dev/"+deviceinfo.DeviceName)
+	if err := cmd1.Run(); err != nil {
+		fmt.Printf("failed to run pvcreate")
+	}
+	cmd2 := exec.Command("vgcreate", deviceinfo.DeviceName, "/dev/"+deviceinfo.DeviceName)
+	if err := cmd2.Run(); err != nil {
+		fmt.Printf("failed to run vgcreate")
+	}
+
+	return nil
+}
+
+func txnHeketiCreateBrick(c transaction.TxnCtx) error {
+
+	return nil
+}