mirrors/ceph-csi
1
0
Fork 0
mirror of https://github.com/ceph/ceph-csi.git synced 2024-12-18 02:50:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

cephfs: validate create volume request

Browse Source 
Also add functionality to generate snap from request
and to get mon and clusterID from the request

Signed-off-by: Humble Chirammal <hchiramm@redhat.com>
This commit is contained in:
Humble Chirammal 2020-08-04 09:33:11 +05:30 committed by mergify[bot]
parent 9c000add29
commit 38d005e4e6
1 changed files with 61 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

61
internal/cephfs/util.go
View File

@ -19,6 +19,7 @@ package cephfs
import (
"context"
"encoding/json"
"errors"
"fmt"
"github.com/ceph/ceph-csi/internal/util"
@ -26,6 +27,7 @@ import (
"github.com/container-storage-interface/spec/lib/go/csi"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
klog "k8s.io/klog/v2"
)
type volumeID string
@ -35,6 +37,7 @@ func execCommandErr(ctx context.Context, program string, args ...string) error {
return err
}
// nolint:unparam // todo:program values has to be revisited later
func execCommandJSON(ctx context.Context, v interface{}, program string, args ...string) error {
stdout, _, err := util.ExecCommand(ctx, program, args...)
if err != nil {
@ -69,6 +72,32 @@ func (cs *ControllerServer) validateCreateVolumeRequest(req *csi.CreateVolumeReq
}
}
if req.VolumeContentSource != nil {
volumeSource := req.VolumeContentSource
switch volumeSource.Type.(type) {
case *csi.VolumeContentSource_Snapshot:
snapshot := req.VolumeContentSource.GetSnapshot()
// CSI spec requires returning NOT_FOUND when the volumeSource is missing/incorrect.
if snapshot == nil {
return status.Error(codes.NotFound, "volume Snapshot cannot be empty")
}
if snapshot.GetSnapshotId() == "" {
return status.Error(codes.NotFound, "volume Snapshot ID cannot be empty")
}
case *csi.VolumeContentSource_Volume:
// CSI spec requires returning NOT_FOUND when the volumeSource is missing/incorrect.
vol := req.VolumeContentSource.GetVolume()
if vol == nil {
return status.Error(codes.NotFound, "volume cannot be empty")
}
if vol.GetVolumeId() == "" {
return status.Error(codes.NotFound, "volume ID cannot be empty")
}
default:
return status.Error(codes.InvalidArgument, "unsupported volume data source")
}
}
return nil
}
@ -97,3 +126,35 @@ func (cs *ControllerServer) validateExpandVolumeRequest(req *csi.ControllerExpan
return nil
}
func genSnapFromOptions(ctx context.Context, req *csi.CreateSnapshotRequest) (snap *cephfsSnapshot, err error) {
cephfsSnap := &cephfsSnapshot{}
cephfsSnap.RequestName = req.GetName()
snapOptions := req.GetParameters()
cephfsSnap.Monitors, cephfsSnap.ClusterID, err = getMonsAndClusterID(ctx, snapOptions)
if err != nil {
return nil, err
}
if namePrefix, ok := snapOptions["snapshotNamePrefix"]; ok {
cephfsSnap.NamePrefix = namePrefix
}
return cephfsSnap, nil
}
func getMonsAndClusterID(ctx context.Context, options map[string]string) (monitors, clusterID string, err error) {
var ok bool
if clusterID, ok = options["clusterID"]; !ok {
err = errors.New("clusterID must be set")
return
}
if monitors, err = util.Mons(csiConfigFile, clusterID); err != nil {
klog.Errorf(util.Log(ctx, "failed getting mons (%s)"), err)
err = fmt.Errorf("failed to fetch monitor list using clusterID (%s): %w", clusterID, err)
return
}
return
}