feedc0de/my-rook-config
1
0
Fork 0
forked from rook/rook
Code Pull Requests Activity

Merge pull request #5509 from thotz/deleteuserorbucket

Browse Source 
ceph: obc clean up resources incase of failure
This commit is contained in:
Sébastien Han
2020-06-11 11:52:10 +02:00
committed by GitHub
parent 609d43d050 96e3d1529e
commit 70cd4e0e4d
2 changed files with 46 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/operator/ceph/object/bucket/provisioner.go
+11 -15
View File
@@ -94,11 +94,13 @@ func (p Provisioner) Provision(options *apibkt.BucketOptions) (*bktv1alpha1.Obje
if err != nil {
err = errors.Wrapf(err, "error creating bucket %q", p.bucketName)
logger.Errorf(err.Error())
p.deleteOBCResource("")
return nil, err
}
_, errCode, err := cephObject.SetQuotaUserBucketMax(p.objectContext, p.cephUserName, maxBuckets)
if errCode > 0 {
p.deleteOBCResource(p.bucketName)
return nil, err
}
logger.Infof("set user %q bucket max to %d", p.cephUserName, maxBuckets)
@@ -131,21 +133,25 @@ func (p Provisioner) Grant(options *apibkt.BucketOptions) (*bktv1alpha1.ObjectBu
// need to quota into -1 for restricting creation of new buckets in rgw
_, _, err = cephObject.SetQuotaUserBucketMax(p.objectContext, p.cephUserName, -1)
if err != nil {
p.deleteOBCResource("")
return nil, err
}
// get the bucket's owner via the bucket metadata
stats, _, err := cephObject.GetBucket(p.objectContext, p.bucketName)
if err != nil {
p.deleteOBCResource("")
return nil, errors.Wrapf(err, "could not get bucket stats (bucket: %s)", p.bucketName)
}
objectUser, _, err := cephObject.GetUser(p.objectContext, stats.Owner)
if err != nil {
p.deleteOBCResource("")
return nil, errors.Wrapf(err, "could not get user (user: %s)", stats.Owner)
}
s3svc, err := NewS3Agent(*objectUser.AccessKey, *objectUser.SecretKey, p.getObjectStoreEndpoint())
if err != nil {
p.deleteOBCResource("")
return nil, err
}
@@ -154,6 +160,7 @@ func (p Provisioner) Grant(options *apibkt.BucketOptions) (*bktv1alpha1.ObjectBu
if err != nil {
if aerr, ok := err.(awserr.Error); ok {
if aerr.Code() != "NoSuchBucketPolicy" {
p.deleteOBCResource("")
return nil, err
}
}
@@ -175,6 +182,7 @@ func (p Provisioner) Grant(options *apibkt.BucketOptions) (*bktv1alpha1.ObjectBu
logger.Infof("PutBucketPolicy output: %v", out)
if err != nil {
p.deleteOBCResource("")
return nil, err
}
// returned ob with connection info
@@ -197,11 +205,7 @@ func (p Provisioner) Delete(ob *bktv1alpha1.ObjectBucket) error {
return err
}
err = p.deleteCephUser(p.cephUserName)
if err != nil {
return err
}
err = p.deleteBucket(p.bucketName)
err = p.deleteOBCResource(p.bucketName)
if err != nil {
return err
}
@@ -288,16 +292,8 @@ func (p Provisioner) Revoke(ob *bktv1alpha1.ObjectBucket) error {
}
}
_, _, err = cephObject.UnlinkUser(p.objectContext, p.cephUserName, p.bucketName)
if err != nil {
return err
}
// finally, delete unlinked user
err = p.deleteCephUser(p.cephUserName)
if err != nil {
return err
}
// finally, delete the user
p.deleteOBCResource("")
return nil
}
pkg/operator/ceph/object/bucket/rgw-handlers.go
+35 -40
View File
@@ -7,26 +7,6 @@ import (
cephObject "github.com/rook/rook/pkg/operator/ceph/object"
)
// The bucket (and objects) could be deleted via an s3 iterator, but since radosgw
// supports the `buket rm` command, this is the approch we'll use.
func (p *Provisioner) deleteBucket(bktName string) error {
// delete bucket with purge option to remove all objects
errCode, err := cephObject.DeleteBucket(p.objectContext, p.bucketName, true)
if errCode == cephObject.RGWErrorNone {
logger.Infof("Bucket %s successfully deleted", bktName)
return nil
}
// opinion: "not found" is not an error
if errCode == cephObject.RGWErrorNotFound {
logger.Infof("Bucket %s does not exist", bktName)
return nil
}
return errors.Wrapf(err, "failed to delete bucket %q: errCode: %d", bktName, errCode)
}
func (p *Provisioner) bucketExists(name string) (bool, error) {
_, errCode, err := cephObject.GetBucket(p.objectContext, name)
if errCode != 0 {
@@ -83,26 +63,6 @@ func (p *Provisioner) createCephUser(username string) (accKey string, secKey str
return *u.AccessKey, *u.SecretKey, nil
}
// Delete the Ceph user associated to the passed-in bucket.
func (p *Provisioner) deleteCephUser(username string) error {
logger.Infof("deleting Ceph user %s for bucket %q", username, p.bucketName)
_, errCode, err := cephObject.DeleteUser(p.objectContext, username)
if errCode == cephObject.RGWErrorNone {
logger.Infof("User %s successfully deleted", username)
return nil
}
// opinion: "not found" is not an error
if errCode == cephObject.RGWErrorNotFound {
logger.Infof("User %s does not exist", username)
return nil
}
return errors.Wrapf(err, "failed to delete user %q: errCode: %d", username, errCode)
}
// returns "" if unable to generate a unique name.
func (p *Provisioner) genUserName() (genName string, err error) {
const (
@@ -122,3 +82,38 @@ func (p *Provisioner) genUserName() (genName string, err error) {
}
return genName, nil
}
// Delete the user and bucket created by OBC with help of radosgw-admin commands
// If delete user failed, error is no longer returned since its permission is
// already revoked and hence user is no longer able to access the bucket
// Empty string is passed for bucketName only if user needs to be removed, ex Revoke()
func (p *Provisioner) deleteOBCResource(bucketName string) error {
logger.Infof("deleting Ceph user %s and bucket %q", p.cephUserName, bucketName)
if len(p.cephUserName) > 0 {
_, errCode, _ := cephObject.DeleteUser(p.objectContext, p.cephUserName)
if errCode == cephObject.RGWErrorNone {
logger.Infof("user %q successfully deleted", p.cephUserName)
} else if errCode == cephObject.RGWErrorNotFound {
// opinion: "not found" is not an error
logger.Infof("user %q does not exist", p.cephUserName)
} else {
logger.Warningf("failed to delete user %q: errCode: %d", p.cephUserName, errCode)
}
}
if len(bucketName) > 0 {
// delete bucket with purge option to remove all objects
errCode, err := cephObject.DeleteBucket(p.objectContext, bucketName, true)
if errCode == cephObject.RGWErrorNone {
logger.Infof("bucket %q successfully deleted", p.bucketName)
} else if errCode == cephObject.RGWErrorNotFound {
// opinion: "not found" is not an error
logger.Infof("bucket %q does not exist", p.bucketName)
} else {
return errors.Wrapf(err, "failed to delete bucket %q: errCode: %d", bucketName, errCode)
}
}
return nil
}