mirrors/ceph-csi
1
0
Fork 0
mirror of https://github.com/ceph/ceph-csi.git synced 2025-06-13 10:33:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

code update

Browse Source
This commit is contained in:
Serguei Bezverkhi
2018-03-06 17:33:57 -05:00
parent e9033989a0
commit 1ff1f8be30
16 changed files with 367 additions and 199 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
pkg/rbd/controllerserver.go
View File

@ -20,12 +20,13 @@ import (
"fmt"
"path"
"github.com/container-storage-interface/spec/lib/go/csi/v0"
"github.com/golang/glog"
"github.com/kubernetes-csi/drivers/pkg/csi-common"
"github.com/pborman/uuid"
"golang.org/x/net/context"
"github.com/container-storage-interface/spec/lib/go/csi"
"github.com/kubernetes-csi/drivers/pkg/csi-common"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
const (
@ -36,17 +37,42 @@ type controllerServer struct {
*csicommon.DefaultControllerServer
}
func GetVersionString(ver *csi.Version) string {
return fmt.Sprintf("%d.%d.%d", ver.Major, ver.Minor, ver.Patch)
}
func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) {
if err := cs.Driver.ValidateControllerServiceRequest(req.Version, csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME); err != nil {
if err := cs.Driver.ValidateControllerServiceRequest(csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME); err != nil {
glog.V(3).Infof("invalid create volume req: %v", req)
return nil, err
}
// Check sanity of request Name, Volume Capabilities
if len(req.Name) == 0 {
return nil, status.Error(codes.InvalidArgument, "Volume Name cannot be empty")
}
if req.VolumeCapabilities == nil {
return nil, status.Error(codes.InvalidArgument, "Volume Capabilities cannot be empty")
}
volOptions, err := getRBDVolumeOptions(req.GetParameters())
// Need to check for already existing volume name, and if found
// check for the requested capacity and already allocated capacity
if exVol, err := getRBDVolumeByName(req.GetName()); err == nil {
// Since err is nil, it means the volume with the same name already exists
// need to check if the size of exisiting volume is the same as in new
// request
if exVol.VolSize >= int64(req.GetCapacityRange().GetRequiredBytes()) {
// exisiting volume is compatible with new request and should be reused.
// TODO (sbezverk) Do I need to make sure that RBD volume still exists?
return &csi.CreateVolumeResponse{
Volume: &csi.Volume{
Id: exVol.VolID,
CapacityBytes: int64(exVol.VolSize),
Attributes: req.GetParameters(),
},
}, nil
}
return nil, status.Error(codes.AlreadyExists, fmt.Sprintf("Volume with the same name: %s but with different size already exist", req.GetName()))
}
// TODO (sbezverk) Last check for not exceeding total storage capacity
rbdVol, err := getRBDVolumeOptions(req.GetParameters())
if err != nil {
return nil, err
}
@ -55,21 +81,23 @@ func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
volName := req.GetName()
uniqueID := uuid.NewUUID().String()
if len(volName) == 0 {
volName = volOptions.Pool + "-dynamic-pvc-" + uniqueID
volName = rbdVol.Pool + "-dynamic-pvc-" + uniqueID
}
volOptions.VolName = volName
rbdVol.VolName = volName
volumeID := "csi-rbd-" + uniqueID
rbdVol.VolID = volumeID
// Volume Size - Default is 1 GiB
volSizeBytes := int64(oneGB)
if req.GetCapacityRange() != nil {
volSizeBytes = int64(req.GetCapacityRange().GetRequiredBytes())
}
rbdVol.VolSize = volSizeBytes
volSizeGB := int(volSizeBytes / 1024 / 1024 / 1024)
// Check if there is already RBD image with requested name
found, _, _ := rbdStatus(volOptions)
found, _, _ := rbdStatus(rbdVol, req.GetControllerCreateSecrets())
if !found {
if err := createRBDImage(volOptions, volSizeGB); err != nil {
if err := createRBDImage(rbdVol, volSizeGB, req.GetControllerCreateSecrets()); err != nil {
if err != nil {
glog.Warningf("failed to create volume: %v", err)
return nil, err
@ -77,12 +105,11 @@ func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
}
glog.V(4).Infof("create volume %s", volName)
}
// Storing volInfo into a persistent file, will need info to delete rbd image
// in ControllerUnpublishVolume
if err := persistVolInfo(volumeID, path.Join(PluginFolder, "controller"), volOptions); err != nil {
// Storing volInfo into a persistent file.
if err := persistVolInfo(volumeID, path.Join(PluginFolder, "controller"), rbdVol); err != nil {
glog.Warningf("rbd: failed to store volInfo with error: %v", err)
}
rbdVolumes[volumeID] = *rbdVol
return &csi.CreateVolumeResponse{
Volume: &csi.Volume{
Id: volumeID,
@ -93,23 +120,21 @@ func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
}
func (cs *controllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVolumeRequest) (*csi.DeleteVolumeResponse, error) {
if err := cs.Driver.ValidateControllerServiceRequest(req.Version, csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME); err != nil {
if err := cs.Driver.ValidateControllerServiceRequest(csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME); err != nil {
glog.Warningf("invalid delete volume req: %v", req)
return nil, err
}
// For now the image get unconditionally deleted, but here retention policy can be checked
volumeID := req.GetVolumeId()
volOptions := &rbdVolumeOptions{}
if err := loadVolInfo(volumeID, path.Join(PluginFolder, "controller"), volOptions); err != nil {
rbdVol := &rbdVolume{}
if err := loadVolInfo(volumeID, path.Join(PluginFolder, "controller"), rbdVol); err != nil {
return nil, err
}
volName := volOptions.VolName
volName := rbdVol.VolName
// Deleting rbd image
glog.V(4).Infof("deleting volume %s", volName)
if err := deleteRBDImage(volOptions); err != nil {
glog.V(3).Infof("failed to delete rbd image: %s/%s with error: %v", volOptions.Pool, volName, err)
if err := deleteRBDImage(rbdVol, req.GetControllerDeleteSecrets()); err != nil {
glog.V(3).Infof("failed to delete rbd image: %s/%s with error: %v", rbdVol.Pool, volName, err)
return nil, err
}
// Removing persistent storage file for the unmapped volume
@ -117,6 +142,7 @@ func (cs *controllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVol
return nil, err
}
delete(rbdVolumes, volumeID)
return &csi.DeleteVolumeResponse{}, nil
}

20
pkg/rbd/nodeserver.go
View File

@ -24,7 +24,7 @@ import (
"github.com/golang/glog"
"golang.org/x/net/context"
"github.com/container-storage-interface/spec/lib/go/csi"
"github.com/container-storage-interface/spec/lib/go/csi/v0"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
@ -67,7 +67,7 @@ func (ns *nodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublis
}
volOptions.VolName = volName
// Mapping RBD image
devicePath, err := attachRBDImage(volOptions)
devicePath, err := attachRBDImage(volOptions, req.GetNodePublishSecrets())
if err != nil {
return nil, err
}
@ -130,3 +130,19 @@ func (ns *nodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpu
return &csi.NodeUnpublishVolumeResponse{}, nil
}
func (ns *nodeServer) NodeStageVolume(
ctx context.Context,
req *csi.NodeStageVolumeRequest) (
*csi.NodeStageVolumeResponse, error) {
return nil, status.Error(codes.Unimplemented, "")
}
func (ns *nodeServer) NodeUnstageVolume(
ctx context.Context,
req *csi.NodeUnstageVolumeRequest) (
*csi.NodeUnstageVolumeResponse, error) {
return nil, status.Error(codes.Unimplemented, "")
}

69
pkg/rbd/rbd.go
View File

@ -17,15 +17,25 @@ limitations under the License.
package rbd
import (
"encoding/json"
"io/ioutil"
"os"
"path"
"strings"
"github.com/golang/glog"
"github.com/container-storage-interface/spec/lib/go/csi"
"github.com/container-storage-interface/spec/lib/go/csi/v0"
"github.com/kubernetes-csi/drivers/pkg/csi-common"
)
// PluginFolder defines the location of rbdplugin
const (
PluginFolder = "/var/lib/kubelet/plugins/rbdplugin"
PluginFolder = "/var/lib/kubelet/plugins/csi-rbdplugin"
// RBDUserID used as a key in credentials map to extract the key which is
// passed be the provisioner, the value od RBDUserID must match to the key used
// in Secret object.
RBDUserID = "admin"
)
type rbd struct {
@ -41,13 +51,54 @@ type rbd struct {
var (
rbdDriver *rbd
version = csi.Version{
Minor: 2,
}
version = "0.2.0"
)
func GetSupportedVersions() []*csi.Version {
return []*csi.Version{&version}
var rbdVolumes map[string]rbdVolume
// Init checks for the persistent volume file and loads all found volumes
// into a memory structure
func init() {
rbdVolumes = map[string]rbdVolume{}
if _, err := os.Stat(path.Join(PluginFolder, "controller")); os.IsNotExist(err) {
glog.Infof("rbd: folder %s not found. Creating... \n", path.Join(PluginFolder, "controller"))
if err := os.Mkdir(path.Join(PluginFolder, "controller"), 0755); err != nil {
glog.Fatalf("Failed to create a controller's volumes folder with error: %v\n", err)
}
return
}
// Since "controller" folder exists, it means the rbdplugin has already been running, it means
// there might be some volumes left, they must be re-inserted into rbdVolumes map
loadExVolumes()
}
// loadExVolumes check for any *.json files in the PluginFolder/controller folder
// and loads then into rbdVolumes map
func loadExVolumes() {
rbdVol := rbdVolume{}
files, err := ioutil.ReadDir(path.Join(PluginFolder, "controller"))
if err != nil {
glog.Infof("rbd: failed to read controller's volumes folder: %s error:%v", path.Join(PluginFolder, "controller"), err)
return
}
for _, f := range files {
if !strings.HasSuffix(f.Name(), ".json") {
continue
}
fp, err := os.Open(path.Join(PluginFolder, "controller", f.Name()))
if err != nil {
glog.Infof("rbd: open file: %s err %%v", f.Name(), err)
continue
}
decoder := json.NewDecoder(fp)
if err = decoder.Decode(&rbdVol); err != nil {
glog.Infof("rbd: decode file: %s err: %v", f.Name(), err)
fp.Close()
continue
}
rbdVolumes[rbdVol.VolID] = rbdVol
}
glog.Infof("rbd: Loaded %d volumes from %s", len(rbdVolumes), path.Join(PluginFolder, "controller"))
}
func GetRBDDriver() *rbd {
@ -73,10 +124,10 @@ func NewNodeServer(d *csicommon.CSIDriver) *nodeServer {
}
func (rbd *rbd) Run(driverName, nodeID, endpoint string) {
glog.Infof("Driver: %v version: %v", driverName, GetVersionString(&version))
glog.Infof("Driver: %v version: %v", driverName, version)
// Initialize default library driver
rbd.driver = csicommon.NewCSIDriver(driverName, &version, GetSupportedVersions(), nodeID)
rbd.driver = csicommon.NewCSIDriver(driverName, version, nodeID)
if rbd.driver == nil {
glog.Fatalln("Failed to initialize CSI Driver.")
}

154
pkg/rbd/rbd_util.go
View File

@ -19,15 +19,16 @@ package rbd
import (
"encoding/json"
"fmt"
"github.com/golang/glog"
"io/ioutil"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/kubernetes/pkg/util/keymutex"
"os"
"os/exec"
"path"
"strings"
"time"
"github.com/golang/glog"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/kubernetes/pkg/util/keymutex"
)
const (
@ -44,22 +45,29 @@ const (
rbdImageWatcherSteps = 10
)
type rbdVolumeOptions struct {
VolName string `json:"volName"`
Monitors string `json:"monitors"`
Pool string `json:"pool"`
AdminID string `json:"adminID"`
AdminSecret string `json:"adminSecret"`
UserID string `json:"userID"`
UserSecret string `json:"userSecret"`
ImageFormat string `json:"imageFormat"`
type rbdVolume struct {
VolName string `json:"volName"`
VolID string `json:"volID"`
Monitors string `json:"monitors"`
Pool string `json:"pool"`
ImageFormat string `json:"imageFormat"`
// TODO (sbezverk) check if it is used and how
ImageFeatures []string `json:"imageFeatures"`
VolSize int64 `json:"volSize"`
}
var attachdetachMutex = keymutex.NewKeyMutex()
func getRBDKey(id string, credentials map[string]string) (string, error) {
if key, ok := credentials[id]; ok {
return key, nil
}
return "", fmt.Errorf("RBD key for ID: %s not found", id)
}
// CreateImage creates a new ceph image with provision and volume options.
func createRBDImage(pOpts *rbdVolumeOptions, volSz int) error {
func createRBDImage(pOpts *rbdVolume, volSz int, credentials map[string]string) error {
var output []byte
var err error
@ -68,12 +76,16 @@ func createRBDImage(pOpts *rbdVolumeOptions, volSz int) error {
image := pOpts.VolName
volSzGB := fmt.Sprintf("%dG", volSz)
if pOpts.ImageFormat == rbdImageFormat2 {
glog.V(4).Infof("rbd: create %s size %s format %s (features: %s) using mon %s, pool %s id %s key %s", image, volSzGB, pOpts.ImageFormat, pOpts.ImageFeatures, mon, pOpts.Pool, pOpts.AdminID, pOpts.AdminSecret)
} else {
glog.V(4).Infof("rbd: create %s size %s format %s using mon %s, pool %s id %s key %s", image, volSzGB, pOpts.ImageFormat, mon, pOpts.Pool, pOpts.AdminID, pOpts.AdminSecret)
key, err := getRBDKey(RBDUserID, credentials)
if err != nil {
return err
}
args := []string{"create", image, "--size", volSzGB, "--pool", pOpts.Pool, "--id", pOpts.AdminID, "-m", mon, "--key=" + pOpts.AdminSecret, "--image-format", pOpts.ImageFormat}
if pOpts.ImageFormat == rbdImageFormat2 {
glog.V(4).Infof("rbd: create %s size %s format %s (features: %s) using mon %s, pool %s id %s key %s", image, volSzGB, pOpts.ImageFormat, pOpts.ImageFeatures, mon, pOpts.Pool, RBDUserID, key)
} else {
glog.V(4).Infof("rbd: create %s size %s format %s using mon %s, pool %s id %s key %s", image, volSzGB, pOpts.ImageFormat, mon, pOpts.Pool, RBDUserID, key)
}
args := []string{"create", image, "--size", volSzGB, "--pool", pOpts.Pool, "--id", RBDUserID, "-m", mon, "--key=" + key, "--image-format", pOpts.ImageFormat}
if pOpts.ImageFormat == rbdImageFormat2 {
// if no image features is provided, it results in empty string
// which disable all RBD image format 2 features as we expected
@ -91,22 +103,20 @@ func createRBDImage(pOpts *rbdVolumeOptions, volSz int) error {
// rbdStatus checks if there is watcher on the image.
// It returns true if there is a watcher onthe image, otherwise returns false.
func rbdStatus(b *rbdVolumeOptions) (bool, string, error) {
func rbdStatus(pOpts *rbdVolume, credentials map[string]string) (bool, string, error) {
var err error
var output string
var cmd []byte
image := b.VolName
image := pOpts.VolName
// If we don't have admin id/secret (e.g. attaching), fallback to user id/secret.
id := b.AdminID
secret := b.AdminSecret
if id == "" {
id = b.UserID
secret = b.UserSecret
key, err := getRBDKey(RBDUserID, credentials)
if err != nil {
return false, "", err
}
glog.V(4).Infof("rbd: status %s using mon %s, pool %s id %s key %s", image, b.Monitors, b.Pool, id, secret)
args := []string{"status", image, "--pool", b.Pool, "-m", b.Monitors, "--id", id, "--key=" + secret}
glog.V(4).Infof("rbd: status %s using mon %s, pool %s id %s key %s", image, pOpts.Monitors, pOpts.Pool, RBDUserID, key)
args := []string{"status", image, "--pool", pOpts.Pool, "-m", pOpts.Monitors, "--id", RBDUserID, "--key=" + key}
cmd, err = execCommand("rbd", args)
output = string(cmd)
@ -133,10 +143,10 @@ func rbdStatus(b *rbdVolumeOptions) (bool, string, error) {
}
// DeleteImage deletes a ceph image with provision and volume options.
func deleteRBDImage(b *rbdVolumeOptions) error {
func deleteRBDImage(pOpts *rbdVolume, credentials map[string]string) error {
var output []byte
image := b.VolName
found, _, err := rbdStatus(b)
image := pOpts.VolName
found, _, err := rbdStatus(pOpts, credentials)
if err != nil {
return err
}
@ -144,15 +154,13 @@ func deleteRBDImage(b *rbdVolumeOptions) error {
glog.Info("rbd is still being used ", image)
return fmt.Errorf("rbd %s is still being used", image)
}
id := b.AdminID
secret := b.AdminSecret
if id == "" {
id = b.UserID
secret = b.UserSecret
key, err := getRBDKey(RBDUserID, credentials)
if err != nil {
return err
}
glog.V(4).Infof("rbd: rm %s using mon %s, pool %s id %s key %s", image, b.Monitors, b.Pool, id, secret)
args := []string{"rm", image, "--pool", b.Pool, "--id", id, "-m", b.Monitors, "--key=" + secret}
glog.V(4).Infof("rbd: rm %s using mon %s, pool %s id %s key %s", image, pOpts.Monitors, pOpts.Pool, RBDUserID, key)
args := []string{"rm", image, "--pool", pOpts.Pool, "--id", RBDUserID, "-m", pOpts.Monitors, "--key=" + key}
output, err = execCommand("rbd", args)
if err == nil {
return nil
@ -166,42 +174,26 @@ func execCommand(command string, args []string) ([]byte, error) {
return cmd.CombinedOutput()
}
func getRBDVolumeOptions(volOptions map[string]string) (*rbdVolumeOptions, error) {
rbdVolume := &rbdVolumeOptions{}
func getRBDVolumeOptions(volOptions map[string]string) (*rbdVolume, error) {
var ok bool
rbdVolume.AdminID, ok = volOptions["adminID"]
if !ok {
return nil, fmt.Errorf("Missing required parameter adminID")
}
rbdVolume.AdminSecret, ok = volOptions["adminSecret"]
if !ok {
return nil, fmt.Errorf("Missing required parameter adminSecret")
}
rbdVolume.Pool, ok = volOptions["pool"]
rbdVol := &rbdVolume{}
rbdVol.Pool, ok = volOptions["pool"]
if !ok {
return nil, fmt.Errorf("Missing required parameter pool")
}
rbdVolume.Monitors, ok = volOptions["monitors"]
rbdVol.Monitors, ok = volOptions["monitors"]
if !ok {
return nil, fmt.Errorf("Missing required parameter monitors")
}
rbdVolume.UserID, ok = volOptions["userID"]
rbdVol.ImageFormat, ok = volOptions["imageFormat"]
if !ok {
return nil, fmt.Errorf("Missing required parameter userID")
}
rbdVolume.UserSecret, ok = volOptions["userSecret"]
if !ok {
return nil, fmt.Errorf("Missing required parameter userSecret")
}
rbdVolume.ImageFormat, ok = volOptions["imageFormat"]
if !ok {
rbdVolume.ImageFormat = "2"
rbdVol.ImageFormat = "2"
}
return rbdVolume, nil
return rbdVol, nil
}
func attachRBDImage(volOptions *rbdVolumeOptions) (string, error) {
func attachRBDImage(volOptions *rbdVolume, credentials map[string]string) (string, error) {
var err error
var output []byte
@ -222,7 +214,7 @@ func attachRBDImage(volOptions *rbdVolumeOptions) (string, error) {
Steps: rbdImageWatcherSteps,
}
err := wait.ExponentialBackoff(backoff, func() (bool, error) {
used, rbdOutput, err := rbdStatus(volOptions)
used, rbdOutput, err := rbdStatus(volOptions, credentials)
if err != nil {
return false, fmt.Errorf("fail to check rbd image status with: (%v), rbd output: (%s)", err, rbdOutput)
}
@ -238,11 +230,13 @@ func attachRBDImage(volOptions *rbdVolumeOptions) (string, error) {
}
glog.V(1).Infof("rbd: map mon %s", volOptions.Monitors)
id := volOptions.UserID
secret := volOptions.UserSecret
key, err := getRBDKey(RBDUserID, credentials)
if err != nil {
return "", err
}
output, err = execCommand("rbd", []string{
"map", image, "--pool", volOptions.Pool, "--id", id, "-m", volOptions.Monitors, "--key=" + secret})
"map", image, "--pool", volOptions.Pool, "--id", RBDUserID, "-m", volOptions.Monitors, "--key=" + key})
if err != nil {
glog.V(1).Infof("rbd: map error %v, rbd output: %s", err, string(output))
return "", fmt.Errorf("rbd: map failed %v, rbd output: %s", err, string(output))
@ -322,23 +316,24 @@ func waitForPath(pool, image string, maxRetries int) (string, bool) {
return "", false
}
func persistVolInfo(image string, persistentStoragePath string, volInfo *rbdVolumeOptions) error {
func persistVolInfo(image string, persistentStoragePath string, volInfo *rbdVolume) error {
file := path.Join(persistentStoragePath, image+".json")
fp, err := os.Create(file)
if err != nil {
glog.Errorf("rbd: failed to create persistent storage file %s with error: %v\n", file, err)
return fmt.Errorf("rbd: create err %s/%s", file, err)
}
defer fp.Close()
encoder := json.NewEncoder(fp)
if err = encoder.Encode(volInfo); err != nil {
return fmt.Errorf("rbd: encode err: %v.", err)
glog.Errorf("rbd: failed to encode volInfo: %+v for file: %s with error: %v\n", volInfo, file, err)
return fmt.Errorf("rbd: encode err: %v", err)
}
glog.Infof("rbd: successfully saved volInfo: %+v into file: %s\n", volInfo, file)
return nil
}
func loadVolInfo(image string, persistentStoragePath string, volInfo *rbdVolumeOptions) error {
func loadVolInfo(image string, persistentStoragePath string, volInfo *rbdVolume) error {
file := path.Join(persistentStoragePath, image+".json")
fp, err := os.Open(file)
if err != nil {
@ -356,11 +351,28 @@ func loadVolInfo(image string, persistentStoragePath string, volInfo *rbdVolumeO
func deleteVolInfo(image string, persistentStoragePath string) error {
file := path.Join(persistentStoragePath, image+".json")
glog.Infof("rbd: Deleting file for Volume: %s at: %s resulting path: %+v\n", image, persistentStoragePath, file)
err := os.Remove(file)
if err != nil {
if err != os.ErrNotExist {
return fmt.Errorf("rbd: open err %s/%s", file, err)
return fmt.Errorf("rbd: error removing file: %s/%s", file, err)
}
}
return nil
}
func getRBDVolumeByID(volumeID string) (rbdVolume, error) {
if rbdVol, ok := rbdVolumes[volumeID]; ok {
return rbdVol, nil
}
return rbdVolume{}, fmt.Errorf("volume id %s does not exit in the volumes list", volumeID)
}
func getRBDVolumeByName(volName string) (rbdVolume, error) {
for _, rbdVol := range rbdVolumes {
if rbdVol.VolName == volName {
return rbdVol, nil
}
}
return rbdVolume{}, fmt.Errorf("volume name %s does not exit in the volumes list", volName)
}