ceph-csi/internal/rbd/clone.go
Niels de Vos 5c26fbb0d7 util: use ClusterConnection.Copy() for re-using connections
Connections are reference counted, so just assigning the connection to
an other object for re-use is not correct. This can cause connections to
be garbage collected while something else is still using it.

Signed-off-by: Niels de Vos <ndevos@redhat.com>
2021-03-17 07:50:09 +00:00

251 lines
8.4 KiB
Go

/*
Copyright 2020 The Ceph-CSI 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 (
"context"
"errors"
"fmt"
"github.com/ceph/ceph-csi/internal/journal"
"github.com/ceph/ceph-csi/internal/util"
librbd "github.com/ceph/go-ceph/rbd"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
// checkCloneImage check the cloned image exists, if the cloned image is not
// found it will check the temporary cloned snapshot exists, and again it will
// check the snapshot exists on the temporary cloned image, if yes it will
// create a new cloned and delete the temporary snapshot and adds a task to
// flatten the temp cloned image and return success.
//
// if the temporary snapshot does not exists it creates a temporary snapshot on
// temporary cloned image and creates a new cloned with user-provided image
// features and delete the temporary snapshot and adds a task to flatten the
// temp cloned image and return success
//
// if the temporary clone does not exist and if there is a temporary snapshot
// present on the parent image it will delete the temporary snapshot and
// returns.
func (rv *rbdVolume) checkCloneImage(ctx context.Context, parentVol *rbdVolume) (bool, error) {
// generate temp cloned volume
tempClone := rv.generateTempClone()
snap := &rbdSnapshot{}
snap.RbdSnapName = rv.RbdImageName
snap.Pool = rv.Pool
// check if cloned image exists
err := rv.getImageInfo()
if err == nil {
// check if do we have temporary snapshot on temporary cloned image
sErr := tempClone.checkSnapExists(snap)
if sErr != nil {
if errors.Is(err, ErrSnapNotFound) {
return true, nil
}
return false, err
}
err = tempClone.deleteSnapshot(ctx, snap)
if err == nil {
return true, nil
}
return false, err
}
if !errors.Is(err, ErrImageNotFound) {
// return error if its not image not found
return false, err
}
err = tempClone.checkSnapExists(snap)
if err != nil {
if errors.Is(err, ErrSnapNotFound) {
// check temporary image needs flatten, if yes add task to flatten the
// temporary clone
err = tempClone.flattenRbdImage(ctx, rv.conn.Creds, false, rbdHardMaxCloneDepth, rbdSoftMaxCloneDepth)
if err != nil {
return false, err
}
// as the snapshot is not present, create new snapshot,clone and
// delete the temporary snapshot
err = createRBDClone(ctx, tempClone, rv, snap, rv.conn.Creds)
if err != nil {
return false, err
}
return true, nil
} else if !errors.Is(err, ErrImageNotFound) {
// any error other than image not found return error
return false, err
}
} else {
// snap will be create after we flatten the temporary cloned image,no
// need to check for flatten here.
// as the snap exists,create clone image and delete temporary snapshot
// and add task to flatten temporary cloned image
err = rv.cloneRbdImageFromSnapshot(ctx, snap)
if err != nil {
util.ErrorLog(ctx, "failed to clone rbd image %s from snapshot %s: %v", rv.RbdImageName, snap.RbdSnapName, err)
err = fmt.Errorf("failed to clone rbd image %s from snapshot %s: %w", rv.RbdImageName, snap.RbdSnapName, err)
return false, err
}
err = tempClone.deleteSnapshot(ctx, snap)
if err != nil {
util.ErrorLog(ctx, "failed to delete snapshot: %v", err)
return false, err
}
return true, nil
}
// as the temp clone does not exist,check snapshot exists on parent volume
// snapshot name is same as temporary clone image
snap.RbdImageName = tempClone.RbdImageName
err = parentVol.checkSnapExists(snap)
if err == nil {
// the temp clone exists, delete it lets reserve a new ID and
// create new resources for clearner approach
err = parentVol.deleteSnapshot(ctx, snap)
}
if errors.Is(err, ErrSnapNotFound) {
return false, nil
}
return false, err
}
func (rv *rbdVolume) generateTempClone() *rbdVolume {
tempClone := rbdVolume{}
tempClone.conn = rv.conn.Copy()
// The temp clone image need to have deep flatten feature
f := []string{librbd.FeatureNameLayering, librbd.FeatureNameDeepFlatten}
tempClone.imageFeatureSet = librbd.FeatureSetFromNames(f)
tempClone.ClusterID = rv.ClusterID
tempClone.Monitors = rv.Monitors
tempClone.Pool = rv.Pool
tempClone.RadosNamespace = rv.RadosNamespace
// The temp cloned image name will be always (rbd image name + "-temp")
// this name will be always unique, as cephcsi never creates an image with
// this format for new rbd images
tempClone.RbdImageName = rv.RbdImageName + "-temp"
return &tempClone
}
func (rv *rbdVolume) createCloneFromImage(ctx context.Context, parentVol *rbdVolume) error {
// generate temp cloned volume
tempClone := rv.generateTempClone()
// snapshot name is same as temporary cloned image, This helps to
// flatten the temporary cloned images as we cannot have more than 510
// snapshots on an rbd image
tempSnap := &rbdSnapshot{}
tempSnap.RbdSnapName = tempClone.RbdImageName
tempSnap.Pool = rv.Pool
cloneSnap := &rbdSnapshot{}
cloneSnap.RbdSnapName = rv.RbdImageName
cloneSnap.Pool = rv.Pool
var (
errClone error
errFlatten error
err error
)
var j = &journal.Connection{}
j, err = volJournal.Connect(rv.Monitors, rv.RadosNamespace, rv.conn.Creds)
if err != nil {
return status.Error(codes.Internal, err.Error())
}
defer j.Destroy()
// create snapshot and temporary clone and delete snapshot
err = createRBDClone(ctx, parentVol, tempClone, tempSnap, rv.conn.Creds)
if err != nil {
return err
}
defer func() {
if err != nil || errClone != nil {
cErr := cleanUpSnapshot(ctx, tempClone, cloneSnap, rv, rv.conn.Creds)
if cErr != nil {
util.ErrorLog(ctx, "failed to cleanup image %s or snapshot %s: %v", cloneSnap, tempClone, cErr)
}
}
if err != nil || errFlatten != nil {
if !errors.Is(errFlatten, ErrFlattenInProgress) {
// cleanup snapshot
cErr := cleanUpSnapshot(ctx, parentVol, tempSnap, tempClone, rv.conn.Creds)
if cErr != nil {
util.ErrorLog(ctx, "failed to cleanup image %s or snapshot %s: %v", tempSnap, tempClone, cErr)
}
}
}
}()
// flatten clone
errFlatten = tempClone.flattenRbdImage(ctx, rv.conn.Creds, false, rbdHardMaxCloneDepth, rbdSoftMaxCloneDepth)
if errFlatten != nil {
return errFlatten
}
// create snap of temp clone from temporary cloned image
// create final clone
// delete snap of temp clone
errClone = createRBDClone(ctx, tempClone, rv, cloneSnap, rv.conn.Creds)
if errClone != nil {
// set errFlatten error to cleanup temporary snapshot and temporary clone
errFlatten = errors.New("failed to create user requested cloned image")
return errClone
}
err = rv.getImageID()
if err != nil {
util.ErrorLog(ctx, "failed to get volume id %s: %v", rv, err)
return err
}
err = j.StoreImageID(ctx, rv.JournalPool, rv.ReservedID, rv.ImageID)
if err != nil {
util.ErrorLog(ctx, "failed to store volume %s: %v", rv, err)
return err
}
return nil
}
func (rv *rbdVolume) flattenCloneImage(ctx context.Context) error {
tempClone := rv.generateTempClone()
// reducing the limit for cloned images to make sure the limit is in range,
// If the intermediate clone reaches the depth we may need to return ABORT
// error message as it need to be flatten before continuing, this may leak
// omap entries and stale temporary snapshots in corner cases, if we reduce
// the limit and check for the depth of the parent image clain it self we
// can flatten the parent images before use to avoid the stale omap entries.
hardLimit := rbdHardMaxCloneDepth
softLimit := rbdSoftMaxCloneDepth
// choosing 2 so that we don't need to flatten the image in the request.
const depthToAvoidFlatten = 2
if rbdHardMaxCloneDepth < depthToAvoidFlatten {
hardLimit = rbdHardMaxCloneDepth - depthToAvoidFlatten
}
if rbdSoftMaxCloneDepth < depthToAvoidFlatten {
softLimit = rbdSoftMaxCloneDepth - depthToAvoidFlatten
}
err := tempClone.getImageInfo()
if err == nil {
return tempClone.flattenRbdImage(ctx, tempClone.conn.Creds, false, hardLimit, softLimit)
}
if !errors.Is(err, ErrImageNotFound) {
return err
}
return rv.flattenRbdImage(ctx, rv.conn.Creds, false, hardLimit, softLimit)
}