gd2 plugin: added a plugin for block volume management

 - added APIs for creation,deleting and listing block volumes.
 - added pluggable interface for block volume providers.

Refer Design Doc: gluster#1319

Signed-off-by: Oshank Kumar <[email protected]>

Author: Oshank Kumar

glusterd2/commands/volumes/volume-create.go

@@ -1,6 +1,7 @@
 package volumecommands
 
 import (
+	"context"
 	"errors"
 	"net/http"
 	"path/filepath"
@@ -99,67 +100,93 @@ func registerVolCreateStepFuncs() {
 }
 
 func volumeCreateHandler(w http.ResponseWriter, r *http.Request) {
+	var (
+		err    error
+		req    api.VolCreateReq
+		ctx    = r.Context()
+		logger = gdctx.GetReqLogger(ctx)
+	)
 
-	ctx := r.Context()
-	ctx, span := trace.StartSpan(ctx, "/volumeCreateHandler")
-	defer span.End()
-
-	logger := gdctx.GetReqLogger(ctx)
-	var err error
-
-	var req api.VolCreateReq
 	if err := restutils.UnmarshalRequest(r, &req); err != nil {
 		restutils.SendHTTPError(ctx, w, http.StatusBadRequest, gderrors.ErrJSONParsingFailed)
 		return
 	}
 
-	if err := validateVolCreateReq(&req); err != nil {
-		restutils.SendHTTPError(ctx, w, http.StatusBadRequest, err)
+	if status, err := CreateVolume(ctx, req); err != nil {
+		restutils.SendHTTPError(ctx, w, status, err)
+		return
+	}
+
+	volinfo, err := volume.GetVolume(req.Name)
+	if err != nil {
+		// FIXME: If volume was created successfully in the txn above and
+		// then the store goes down by the time we reach here, what do
+		// we return to the client ?
+		restutils.SendHTTPError(ctx, w, http.StatusInternalServerError, err)
 		return
 	}
 
+	logger.WithField("volume-name", volinfo.Name).Info("new volume created")
+	events.Broadcast(volume.NewEvent(volume.EventVolumeCreated, volinfo))
+
+	resp := createVolumeCreateResp(volinfo)
+	restutils.SetLocationHeader(r, w, volinfo.Name)
+	restutils.SendHTTPResponse(ctx, w, http.StatusCreated, resp)
+}
+
+func createVolumeCreateResp(v *volume.Volinfo) *api.VolumeCreateResp {
+	return (*api.VolumeCreateResp)(volume.CreateVolumeInfoResp(v))
+}
+
+// CreateVolume will create a volume. It returns http StatusCode to be sent to client
+// and any error if occurred.
+func CreateVolume(ctx context.Context, req api.VolCreateReq) (status int, err error) {
+	ctx, span := trace.StartSpan(ctx, "/volumeCreateHandler")
+	defer span.End()
+
+	if err := validateVolCreateReq(&req); err != nil {
+		return http.StatusBadRequest, err
+	}
+
 	if req.Size > 0 {
 		applyDefaults(&req)
 
 		if req.SnapshotReserveFactor < 1 {
-			restutils.SendHTTPError(ctx, w, http.StatusBadRequest,
-				errors.New("invalid snapshot reserve factor"))
-			return
+			return http.StatusBadRequest, errors.New("invalid snapshot reserve factor")
+
 		}
 
 		if err := bricksplanner.PlanBricks(&req); err != nil {
-			restutils.SendHTTPError(ctx, w, http.StatusInternalServerError, err)
-			return
+			return http.StatusInternalServerError, err
+
 		}
 	} else {
 		if err := checkDupBrickEntryVolCreate(req); err != nil {
-			restutils.SendHTTPError(ctx, w, http.StatusBadRequest, err)
-			return
+			return http.StatusBadRequest, err
+
 		}
 	}
 
 	req.Options, err = expandGroupOptions(req.Options)
 	if err != nil {
-		restutils.SendHTTPError(ctx, w, http.StatusInternalServerError, err)
-		return
+		return http.StatusInternalServerError, err
+
 	}
 
 	if err := validateOptions(req.Options, req.VolOptionFlags); err != nil {
-		restutils.SendHTTPError(ctx, w, http.StatusBadRequest, err)
-		return
+		return http.StatusBadRequest, err
+
 	}
 
 	nodes, err := req.Nodes()
 	if err != nil {
-		restutils.SendHTTPError(ctx, w, http.StatusBadRequest, err)
-		return
+		return http.StatusBadRequest, err
+
 	}
 
 	txn, err := transaction.NewTxnWithLocks(ctx, req.Name)
 	if err != nil {
-		status, err := restutils.ErrToStatusCode(err)
-		restutils.SendHTTPError(ctx, w, status, err)
-		return
+		return restutils.ErrToStatusCode(err)
 	}
 	defer txn.Done()
 
@@ -191,8 +218,8 @@ func volumeCreateHandler(w http.ResponseWriter, r *http.Request) {
 	}
 
 	if err := txn.Ctx.Set("req", &req); err != nil {
-		restutils.SendHTTPError(ctx, w, http.StatusInternalServerError, err)
-		return
+		return http.StatusInternalServerError, err
+
 	}
 
 	// Add attributes to the span with info that can be viewed along with traces.
@@ -203,28 +230,8 @@ func volumeCreateHandler(w http.ResponseWriter, r *http.Request) {
 	)
 
 	if err := txn.Do(); err != nil {
-		status, err := restutils.ErrToStatusCode(err)
-		restutils.SendHTTPError(ctx, w, status, err)
-		return
+		return restutils.ErrToStatusCode(err)
 	}
 
-	volinfo, err := volume.GetVolume(req.Name)
-	if err != nil {
-		// FIXME: If volume was created successfully in the txn above and
-		// then the store goes down by the time we reach here, what do
-		// we return to the client ?
-		restutils.SendHTTPError(ctx, w, http.StatusInternalServerError, err)
-		return
-	}
-
-	logger.WithField("volume-name", volinfo.Name).Info("new volume created")
-	events.Broadcast(volume.NewEvent(volume.EventVolumeCreated, volinfo))
-
-	resp := createVolumeCreateResp(volinfo)
-	restutils.SetLocationHeader(r, w, volinfo.Name)
-	restutils.SendHTTPResponse(ctx, w, http.StatusCreated, resp)
-}
-
-func createVolumeCreateResp(v *volume.Volinfo) *api.VolumeCreateResp {
-	return (*api.VolumeCreateResp)(volume.CreateVolumeInfoResp(v))
+	return http.StatusCreated, nil
 }

glusterd2/plugin/plugins.go

@@ -4,6 +4,7 @@ package plugin
 
 import (
 	"github.com/gluster/glusterd2/plugins/bitrot"
+	"github.com/gluster/glusterd2/plugins/blockvolume"
 	"github.com/gluster/glusterd2/plugins/device"
 	"github.com/gluster/glusterd2/plugins/events"
 	"github.com/gluster/glusterd2/plugins/georeplication"
@@ -25,4 +26,5 @@ var PluginsList = []GlusterdPlugin{
 	&glustershd.Plugin{},
 	&device.Plugin{},
 	&rebalance.Plugin{},
+	&blockvolume.BlockVolume{},
 }

glusterd2/volume/filters.go

@@ -0,0 +1,51 @@
+package volume
+
+const (
+	// BlockHosted is plugin name for FilterBlockHostedVolumes
+	BlockHosted = "block-hosted"
+)
+
+// Filter will receive a slice of *Volinfo and filters out the undesired one and return slice of desired one only
+type Filter func([]*Volinfo) []*Volinfo
+
+var filters = make(map[string]Filter)
+
+// InstallFilter will register a custom Filter
+func InstallFilter(name string, f Filter) {
+	filters[name] = f
+}
+
+// ApplyFilters applies all registered filters passed in the args to a slice of *Volinfo
+func ApplyFilters(volumes []*Volinfo, names ...string) []*Volinfo {
+	for _, name := range names {
+		if filter, found := filters[name]; found {
+			volumes = filter(volumes)
+		}
+	}
+	return volumes
+}
+
+// ApplyCustomFilters applies all custom filter to a slice of *Volinfo
+func ApplyCustomFilters(volumes []*Volinfo, filters ...Filter) []*Volinfo {
+	for _, filter := range filters {
+		volumes = filter(volumes)
+	}
+
+	return volumes
+}
+
+// FilterBlockHostedVolumes filters out volume which are suitable for hosting block volume
+func FilterBlockHostedVolumes(volumes []*Volinfo) []*Volinfo {
+	var volInfos []*Volinfo
+	for _, volume := range volumes {
+		val, found := volume.Metadata["_block-hosting"]
+		if found && val == "yes" {
+			volInfos = append(volInfos, volume)
+		}
+	}
+	return volInfos
+}
+
+func init() {
+	InstallFilter(BlockHosted, FilterBlockHostedVolumes)
+}