ceph-csi/pkg/rbd/nodeserver.go

296 lines
9.3 KiB
Go
Raw Normal View History

2018-01-09 18:59:50 +00:00
/*
Copyright 2018 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package rbd
import (
"fmt"
2018-01-09 18:59:50 +00:00
"os"
"os/exec"
"regexp"
"strings"
2018-01-09 18:59:50 +00:00
"github.com/golang/glog"
"golang.org/x/net/context"
"github.com/container-storage-interface/spec/lib/go/csi"
2018-01-09 18:59:50 +00:00
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
2018-01-09 18:59:50 +00:00
"k8s.io/kubernetes/pkg/util/mount"
"github.com/kubernetes-csi/drivers/pkg/csi-common"
)
// NodeServer struct of ceph rbd driver with supported methods of CSI
// node server spec
type NodeServer struct {
2018-01-09 18:59:50 +00:00
*csicommon.DefaultNodeServer
mounter mount.Interface
2018-01-09 18:59:50 +00:00
}
//TODO remove both stage and unstage methods
//once https://github.com/kubernetes-csi/drivers/pull/145 is merged
// NodeStageVolume returns unimplemented response
func (ns *NodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRequest) (*csi.NodeStageVolumeResponse, error) {
return nil, status.Error(codes.Unimplemented, "")
}
// NodeUnstageVolume returns unimplemented response
func (ns *NodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUnstageVolumeRequest) (*csi.NodeUnstageVolumeResponse, error) {
return nil, status.Error(codes.Unimplemented, "")
}
// NodePublishVolume mounts the volume mounted to the device path to the target
// path
func (ns *NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolumeRequest) (*csi.NodePublishVolumeResponse, error) {
2018-01-09 18:59:50 +00:00
targetPath := req.GetTargetPath()
targetPathMutex.LockKey(targetPath)
defer func() {
if err := targetPathMutex.UnlockKey(targetPath); err != nil {
glog.Warningf("failed to unlock mutex targetpath:%s %v", targetPath, err)
}
}()
volName, err := ns.getVolumeName(req)
if err != nil {
return nil, err
}
2018-11-01 01:03:03 +00:00
isBlock := req.GetVolumeCapability().GetBlock() != nil
// Check if that target path exists properly
notMnt, err := ns.createTargetPath(targetPath, isBlock)
if err != nil {
return nil, err
}
2018-01-09 18:59:50 +00:00
if !notMnt {
return &csi.NodePublishVolumeResponse{}, nil
2018-01-09 18:59:50 +00:00
}
volOptions, err := getRBDVolumeOptions(req.GetVolumeContext())
2018-01-09 18:59:50 +00:00
if err != nil {
return nil, err
}
volOptions.VolName = volName
2018-01-09 18:59:50 +00:00
// Mapping RBD image
devicePath, err := attachRBDImage(volOptions, volOptions.UserID, req.GetSecrets())
2018-01-09 18:59:50 +00:00
if err != nil {
return nil, err
}
2018-09-21 14:07:14 +00:00
glog.V(4).Infof("rbd image: %s/%s was successfully mapped at %s\n", req.GetVolumeId(), volOptions.Pool, devicePath)
2018-01-09 18:59:50 +00:00
// Publish Path
err = ns.mountVolume(req, devicePath)
if err != nil {
return nil, err
}
return &csi.NodePublishVolumeResponse{}, nil
}
func (ns *NodeServer) getVolumeName(req *csi.NodePublishVolumeRequest) (string, error) {
var volName string
isBlock := req.GetVolumeCapability().GetBlock() != nil
targetPath := req.GetTargetPath()
if isBlock {
// Get volName from targetPath
s := strings.Split(targetPath, "/")
volName = s[len(s)-1]
} else {
// Get volName from targetPath
if !strings.HasSuffix(targetPath, "/mount") {
return "", fmt.Errorf("rbd: malformed the value of target path: %s", targetPath)
}
s := strings.Split(strings.TrimSuffix(targetPath, "/mount"), "/")
volName = s[len(s)-1]
}
return volName, nil
}
func (ns *NodeServer) mountVolume(req *csi.NodePublishVolumeRequest, devicePath string) error {
2018-11-01 01:03:03 +00:00
// Publish Path
fsType := req.GetVolumeCapability().GetMount().GetFsType()
2018-01-09 18:59:50 +00:00
readOnly := req.GetReadonly()
attrib := req.GetVolumeContext()
2018-01-09 18:59:50 +00:00
mountFlags := req.GetVolumeCapability().GetMount().GetMountFlags()
isBlock := req.GetVolumeCapability().GetBlock() != nil
targetPath := req.GetTargetPath()
2018-01-09 18:59:50 +00:00
2018-11-01 01:03:03 +00:00
glog.V(4).Infof("target %v\nisBlock %v\nfstype %v\ndevice %v\nreadonly %v\nattributes %v\n mountflags %v\n",
targetPath, isBlock, fsType, devicePath, readOnly, attrib, mountFlags)
2018-01-09 18:59:50 +00:00
diskMounter := &mount.SafeFormatAndMount{Interface: ns.mounter, Exec: mount.NewOsExec()}
2018-11-01 01:03:03 +00:00
if isBlock {
options := []string{"bind"}
if err := diskMounter.Mount(devicePath, targetPath, fsType, options); err != nil {
return err
2018-11-01 01:03:03 +00:00
}
} else {
options := []string{}
if readOnly {
options = append(options, "ro")
}
if err := diskMounter.FormatAndMount(devicePath, targetPath, fsType, options); err != nil {
return err
2018-11-01 01:03:03 +00:00
}
2018-01-09 18:59:50 +00:00
}
return nil
}
func (ns *NodeServer) createTargetPath(targetPath string, isBlock bool) (bool, error) {
// Check if that target path exists properly
notMnt, err := ns.mounter.IsNotMountPoint(targetPath)
if err != nil {
if os.IsNotExist(err) {
if isBlock {
// create an empty file
// #nosec
targetPathFile, e := os.OpenFile(targetPath, os.O_CREATE|os.O_RDWR, 0750)
if e != nil {
glog.V(4).Infof("Failed to create targetPath:%s with error: %v", targetPath, err)
return notMnt, status.Error(codes.Internal, e.Error())
}
if err = targetPathFile.Close(); err != nil {
glog.V(4).Infof("Failed to close targetPath:%s with error: %v", targetPath, err)
return notMnt, status.Error(codes.Internal, err.Error())
}
} else {
// Create a directory
if err = os.MkdirAll(targetPath, 0750); err != nil {
return notMnt, status.Error(codes.Internal, err.Error())
}
}
notMnt = true
} else {
return false, status.Error(codes.Internal, err.Error())
}
}
return notMnt, err
2018-01-09 18:59:50 +00:00
}
// NodeUnpublishVolume unmounts the volume from the target path
func (ns *NodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpublishVolumeRequest) (*csi.NodeUnpublishVolumeResponse, error) {
2018-01-09 18:59:50 +00:00
targetPath := req.GetTargetPath()
targetPathMutex.LockKey(targetPath)
defer func() {
if err := targetPathMutex.UnlockKey(targetPath); err != nil {
glog.Warningf("failed to unlock mutex targetpath:%s %v", targetPath, err)
}
}()
2018-01-09 18:59:50 +00:00
notMnt, err := ns.mounter.IsNotMountPoint(targetPath)
if err != nil {
if os.IsNotExist(err) {
// targetPath has already been deleted
glog.V(4).Infof("targetPath: %s has already been deleted", targetPath)
return &csi.NodeUnpublishVolumeResponse{}, nil
}
return nil, status.Error(codes.NotFound, err.Error())
}
if notMnt {
// TODO should consider deleting path instead of returning error,
// once all codes become ready for csi 1.0.
return nil, status.Error(codes.NotFound, "Volume not mounted")
}
devicePath, cnt, err := mount.GetDeviceNameFromMount(ns.mounter, targetPath)
if err != nil {
return nil, status.Error(codes.Internal, err.Error())
}
if err = ns.unmount(targetPath, devicePath, cnt); err != nil {
return nil, err
}
return &csi.NodeUnpublishVolumeResponse{}, nil
}
func (ns *NodeServer) unmount(targetPath, devicePath string, cnt int) error {
var err error
// Bind mounted device needs to be resolved by using resolveBindMountedBlockDevice
if devicePath == "devtmpfs" {
devicePath, err = resolveBindMountedBlockDevice(targetPath)
if err != nil {
return status.Error(codes.Internal, err.Error())
}
glog.V(4).Infof("NodeUnpublishVolume: devicePath: %s, (original)cnt: %d\n", devicePath, cnt)
// cnt for GetDeviceNameFromMount is broken for bind mouted device,
// it counts total number of mounted "devtmpfs", instead of counting this device.
// So, forcibly setting cnt to 1 here.
// TODO : fix this properly
cnt = 1
}
glog.V(4).Infof("NodeUnpublishVolume: targetPath: %s, devicePath: %s\n", targetPath, devicePath)
2018-01-09 18:59:50 +00:00
// Unmounting the image
err = ns.mounter.Unmount(targetPath)
2018-01-09 18:59:50 +00:00
if err != nil {
glog.V(3).Infof("failed to unmount targetPath: %s with error: %v", targetPath, err)
return status.Error(codes.Internal, err.Error())
2018-01-09 18:59:50 +00:00
}
cnt--
if cnt != 0 {
// TODO should this be fixed not to success, so that driver can retry unmounting?
return nil
2018-01-09 18:59:50 +00:00
}
// Unmapping rbd device
if err = detachRBDDevice(devicePath); err != nil {
glog.V(3).Infof("failed to unmap rbd device: %s with error: %v", devicePath, err)
return err
2018-01-09 18:59:50 +00:00
}
// Remove targetPath
if err = os.RemoveAll(targetPath); err != nil {
glog.V(3).Infof("failed to remove targetPath: %s with error: %v", targetPath, err)
}
return err
2018-01-09 18:59:50 +00:00
}
func resolveBindMountedBlockDevice(mountPath string) (string, error) {
// #nosec
cmd := exec.Command("findmnt", "-n", "-o", "SOURCE", "--first-only", "--target", mountPath)
out, err := cmd.CombinedOutput()
if err != nil {
glog.V(2).Infof("Failed findmnt command for path %s: %s %v", mountPath, out, err)
return "", err
}
return parseFindMntResolveSource(string(out))
}
// parse output of "findmnt -o SOURCE --first-only --target" and return just the SOURCE
func parseFindMntResolveSource(out string) (string, error) {
// cut trailing newline
out = strings.TrimSuffix(out, "\n")
// Check if out is a mounted device
reMnt := regexp.MustCompile("^(/[^/]+(?:/[^/]*)*)$")
if match := reMnt.FindStringSubmatch(out); match != nil {
return match[1], nil
}
// Check if out is a block device
// nolint
reBlk := regexp.MustCompile("^devtmpfs\\[(/[^/]+(?:/[^/]*)*)\\]$")
if match := reBlk.FindStringSubmatch(out); match != nil {
return fmt.Sprintf("/dev%s", match[1]), nil
}
return "", fmt.Errorf("parseFindMntResolveSource: %s doesn't match to any expected findMnt output", out)
}