mirror of
https://github.com/ceph/ceph-csi.git
synced 2024-11-17 20:00:23 +00:00
cephfs/volume: create/delete-volume idempotency checks
This commit is contained in:
gman
parent
5c6bf5fa1a
commit
60588d8968
@ -17,11 +17,11 @@ limitations under the License.
|
||||
package cephfs
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"os"
|
||||
"os/exec"
|
||||
|
||||
"google.golang.org/grpc/codes"
|
||||
@ -30,61 +30,41 @@ import (
|
||||
|
||||
"github.com/ceph/ceph-csi/pkg/util"
|
||||
"github.com/container-storage-interface/spec/lib/go/csi"
|
||||
"k8s.io/kubernetes/pkg/util/keymutex"
|
||||
"k8s.io/kubernetes/pkg/util/mount"
|
||||
)
|
||||
|
||||
type volumeID string
|
||||
|
||||
func mustUnlock(m keymutex.KeyMutex, key string) {
|
||||
if err := m.UnlockKey(key); err != nil {
|
||||
klog.Fatalf("failed to unlock mutex for %s: %v", key, err)
|
||||
}
|
||||
}
|
||||
|
||||
func makeVolumeID(volName string) volumeID {
|
||||
return volumeID("csi-cephfs-" + volName)
|
||||
}
|
||||
|
||||
func closePipeOnError(pipe io.Closer, err error) {
|
||||
if err != nil {
|
||||
if err = pipe.Close(); err != nil {
|
||||
klog.Warningf("failed to close pipe: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func execCommand(program string, args ...string) (stdout, stderr []byte, err error) {
|
||||
cmd := exec.Command(program, args...) // nolint: gosec
|
||||
stripArgs := util.StripSecretInArgs(args)
|
||||
klog.V(4).Infof("cephfs: EXEC %s %s", program, stripArgs)
|
||||
var (
|
||||
cmd = exec.Command(program, args...) // nolint: gosec
|
||||
sanitizedArgs = util.StripSecretInArgs(args)
|
||||
stdoutBuf bytes.Buffer
|
||||
stderrBuf bytes.Buffer
|
||||
)
|
||||
|
||||
stdoutPipe, err := cmd.StdoutPipe()
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("cannot open stdout pipe for %s %v: %v", program, stripArgs, err)
|
||||
cmd.Stdout = &stdoutBuf
|
||||
cmd.Stderr = &stderrBuf
|
||||
|
||||
klog.V(4).Infof("cephfs: EXEC %s %s", program, sanitizedArgs)
|
||||
|
||||
if err := cmd.Run(); err != nil {
|
||||
return nil, nil, fmt.Errorf("an error occurred while running (%d) %s %v: %v: %s",
|
||||
cmd.Process.Pid, program, sanitizedArgs, err, stderrBuf.Bytes())
|
||||
}
|
||||
|
||||
defer closePipeOnError(stdoutPipe, err)
|
||||
|
||||
stderrPipe, err := cmd.StderrPipe()
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("cannot open stdout pipe for %s %v: %v", program, stripArgs, err)
|
||||
}
|
||||
|
||||
defer closePipeOnError(stderrPipe, err)
|
||||
|
||||
if err = cmd.Start(); err != nil {
|
||||
return nil, nil, fmt.Errorf("failed to run %s %v: %v", program, stripArgs, err)
|
||||
}
|
||||
|
||||
stdout, err = ioutil.ReadAll(stdoutPipe)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("failed to read from stdout for %s %v: %v", program, stripArgs, err)
|
||||
}
|
||||
|
||||
stderr, err = ioutil.ReadAll(stderrPipe)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("failed to read from stderr for %s %v: %v", program, stripArgs, err)
|
||||
}
|
||||
|
||||
if waitErr := cmd.Wait(); waitErr != nil {
|
||||
return nil, nil, fmt.Errorf("an error occurred while running %s %v: %v: %s", program, stripArgs, waitErr, stderr)
|
||||
}
|
||||
|
||||
return
|
||||
return stdoutBuf.Bytes(), stderrBuf.Bytes(), nil
|
||||
}
|
||||
|
||||
func execCommandErr(program string, args ...string) error {
|
||||
@ -117,6 +97,11 @@ func isMountPoint(p string) (bool, error) {
|
||||
return !notMnt, nil
|
||||
}
|
||||
|
||||
func pathExists(p string) bool {
|
||||
_, err := os.Stat(p)
|
||||
return err == nil
|
||||
}
|
||||
|
||||
// Controller service request validation
|
||||
func (cs *ControllerServer) validateCreateVolumeRequest(req *csi.CreateVolumeRequest) error {
|
||||
if err := cs.Driver.ValidateControllerServiceRequest(csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME); err != nil {
|
||||
|
||||
@ -52,79 +52,66 @@ func setVolumeAttribute(root, attrName, attrValue string) error {
|
||||
}
|
||||
|
||||
func createVolume(volOptions *volumeOptions, adminCr *credentials, volID volumeID, bytesQuota int64) error {
|
||||
cephRoot := getCephRootPathLocal(volID)
|
||||
|
||||
if err := createMountPoint(cephRoot); err != nil {
|
||||
if err := mountCephRoot(volID, volOptions, adminCr); err != nil {
|
||||
return err
|
||||
}
|
||||
defer unmountCephRoot(volID)
|
||||
|
||||
// RootPath is not set for a dynamically provisioned volume
|
||||
// Access to cephfs's / is required
|
||||
volOptions.RootPath = "/"
|
||||
var (
|
||||
volRoot = getCephRootVolumePathLocal(volID)
|
||||
volRootCreating = volRoot + "-creating"
|
||||
)
|
||||
|
||||
m, err := newMounter(volOptions)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to create mounter: %v", err)
|
||||
if pathExists(volRoot) {
|
||||
klog.V(4).Infof("cephfs: volume %s already exists, skipping creation", volID)
|
||||
return nil
|
||||
}
|
||||
|
||||
if err = m.mount(cephRoot, adminCr, volOptions, volID); err != nil {
|
||||
return fmt.Errorf("error mounting ceph root: %v", err)
|
||||
}
|
||||
|
||||
defer unmountAndRemove(cephRoot)
|
||||
|
||||
volOptions.RootPath = getVolumeRootPathCeph(volID)
|
||||
localVolRoot := getCephRootVolumePathLocal(volID)
|
||||
|
||||
if err := createMountPoint(localVolRoot); err != nil {
|
||||
if err := createMountPoint(volRootCreating); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if bytesQuota > 0 {
|
||||
if err := setVolumeAttribute(localVolRoot, "ceph.quota.max_bytes", fmt.Sprintf("%d", bytesQuota)); err != nil {
|
||||
if err := setVolumeAttribute(volRootCreating, "ceph.quota.max_bytes", fmt.Sprintf("%d", bytesQuota)); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
if err := setVolumeAttribute(localVolRoot, "ceph.dir.layout.pool", volOptions.Pool); err != nil {
|
||||
if err := setVolumeAttribute(volRootCreating, "ceph.dir.layout.pool", volOptions.Pool); err != nil {
|
||||
return fmt.Errorf("%v\ncephfs: Does pool '%s' exist?", err, volOptions.Pool)
|
||||
}
|
||||
|
||||
if err := setVolumeAttribute(localVolRoot, "ceph.dir.layout.pool_namespace", getVolumeNamespace(volID)); err != nil {
|
||||
if err := setVolumeAttribute(volRootCreating, "ceph.dir.layout.pool_namespace", getVolumeNamespace(volID)); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := os.Rename(volRootCreating, volRoot); err != nil {
|
||||
return fmt.Errorf("couldn't mark volume %s as created: %v", volID, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func purgeVolume(volID volumeID, adminCr *credentials, volOptions *volumeOptions) error {
|
||||
if err := mountCephRoot(volID, volOptions, adminCr); err != nil {
|
||||
return err
|
||||
}
|
||||
defer unmountCephRoot(volID)
|
||||
|
||||
var (
|
||||
cephRoot = getCephRootPathLocal(volID)
|
||||
volRoot = getCephRootVolumePathLocal(volID)
|
||||
volRootDeleting = volRoot + "-deleting"
|
||||
)
|
||||
|
||||
if err := createMountPoint(cephRoot); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Root path is not set for dynamically provisioned volumes
|
||||
// Access to cephfs's / is required
|
||||
volOptions.RootPath = "/"
|
||||
|
||||
m, err := newMounter(volOptions)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to create mounter: %v", err)
|
||||
}
|
||||
|
||||
if err = m.mount(cephRoot, adminCr, volOptions, volID); err != nil {
|
||||
return fmt.Errorf("error mounting ceph root: %v", err)
|
||||
}
|
||||
|
||||
defer unmountAndRemove(cephRoot)
|
||||
|
||||
if err := os.Rename(volRoot, volRootDeleting); err != nil {
|
||||
return fmt.Errorf("couldn't mark volume %s for deletion: %v", volID, err)
|
||||
if pathExists(volRoot) {
|
||||
if err := os.Rename(volRoot, volRootDeleting); err != nil {
|
||||
return fmt.Errorf("couldn't mark volume %s for deletion: %v", volID, err)
|
||||
}
|
||||
} else {
|
||||
if !pathExists(volRootDeleting) {
|
||||
klog.V(4).Infof("cephfs: volume %s not found, assuming it to be already deleted", volID)
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
if err := os.RemoveAll(volRootDeleting); err != nil {
|
||||
@ -134,13 +121,37 @@ func purgeVolume(volID volumeID, adminCr *credentials, volOptions *volumeOptions
|
||||
return nil
|
||||
}
|
||||
|
||||
func unmountAndRemove(mountPoint string) {
|
||||
var err error
|
||||
if err = unmountVolume(mountPoint); err != nil {
|
||||
klog.Errorf("failed to unmount %s with error %s", mountPoint, err)
|
||||
func mountCephRoot(volID volumeID, volOptions *volumeOptions, adminCr *credentials) error {
|
||||
cephRoot := getCephRootPathLocal(volID)
|
||||
|
||||
// Root path is not set for dynamically provisioned volumes
|
||||
// Access to cephfs's / is required
|
||||
volOptions.RootPath = "/"
|
||||
|
||||
if err := createMountPoint(cephRoot); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err = os.Remove(mountPoint); err != nil {
|
||||
klog.Errorf("failed to remove %s with error %s", mountPoint, err)
|
||||
m, err := newMounter(volOptions)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to create mounter: %v", err)
|
||||
}
|
||||
|
||||
if err = m.mount(cephRoot, adminCr, volOptions, volID); err != nil {
|
||||
return fmt.Errorf("error mounting ceph root: %v", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func unmountCephRoot(volID volumeID) {
|
||||
cephRoot := getCephRootPathLocal(volID)
|
||||
|
||||
if err := unmountVolume(cephRoot); err != nil {
|
||||
klog.Errorf("failed to unmount %s with error %s", cephRoot, err)
|
||||
}
|
||||
|
||||
if err := os.Remove(cephRoot); err != nil {
|
||||
klog.Errorf("failed to remove %s with error %s", cephRoot, err)
|
||||
}
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user