swift: fix deletion of parts of Static Large Object (SLO)

Before this change, deleting SLO objects could leave the parts of the object behind.
This commit is contained in:
Nguyễn Hữu Luân 2020-12-28 20:21:11 +07:00 committed by GitHub
parent f347a198f7
commit 6342499c47
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

View File

@ -976,6 +976,18 @@ func (o *Object) isStaticLargeObject() (bool, error) {
return o.hasHeader("X-Static-Large-Object") return o.hasHeader("X-Static-Large-Object")
} }
func (o *Object) isLargeObject() (result bool, err error) {
result, err = o.hasHeader("X-Static-Large-Object")
if result {
return
}
result, err = o.hasHeader("X-Object-Manifest")
if result {
return
}
return false, nil
}
func (o *Object) isInContainerVersioning(container string) (bool, error) { func (o *Object) isInContainerVersioning(container string) (bool, error) {
_, headers, err := o.fs.c.Container(container) _, headers, err := o.fs.c.Container(container)
if err != nil { if err != nil {
@ -1364,6 +1376,32 @@ func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, op
func (o *Object) Remove(ctx context.Context) (err error) { func (o *Object) Remove(ctx context.Context) (err error) {
container, containerPath := o.split() container, containerPath := o.split()
//check object is large object
isLargeObject, err := o.isLargeObject()
if err != nil {
return err
}
//check container has enabled version to reserve segment when delete
isInContainerVersioning := false
if isLargeObject {
isInContainerVersioning, err = o.isInContainerVersioning(container)
if err != nil {
return err
}
}
isStaticLargeObject, err := o.isStaticLargeObject()
if err != nil {
return err
}
var segmentContainer string
var segmentObjects []swift.Object
if isStaticLargeObject {
segmentContainer, segmentObjects, err = o.fs.c.LargeObjectGetSegments(container, containerPath)
if err != nil {
return err
}
}
// Remove file/manifest first // Remove file/manifest first
err = o.fs.pacer.Call(func() (bool, error) { err = o.fs.pacer.Call(func() (bool, error) {
err = o.fs.c.ObjectDelete(container, containerPath) err = o.fs.c.ObjectDelete(container, containerPath)
@ -1372,23 +1410,36 @@ func (o *Object) Remove(ctx context.Context) (err error) {
if err != nil { if err != nil {
return err return err
} }
if !isLargeObject || isInContainerVersioning {
return nil
}
isDynamicLargeObject, err := o.isDynamicLargeObject() isDynamicLargeObject, err := o.isDynamicLargeObject()
if err != nil { if err != nil {
return err return err
} }
// ...then segments if required // ...then segments if required
//delete segment for dynamic large object
if isDynamicLargeObject { if isDynamicLargeObject {
isInContainerVersioning, err := o.isInContainerVersioning(container) return o.removeSegments("")
if err != nil {
return err
} }
if !isInContainerVersioning {
err = o.removeSegments("") //delete segment for static large object
if isStaticLargeObject && len(segmentContainer) > 0 && segmentObjects != nil && len(segmentObjects) > 0 {
var segmentNames []string
for _, segmentObject := range segmentObjects {
if len(segmentObject.Name) == 0 {
continue
}
segmentNames = append(segmentNames, segmentObject.Name)
}
_, err := o.fs.c.BulkDelete(segmentContainer, segmentNames)
if err != nil { if err != nil {
return err return err
} }
} }
}
return nil return nil
} }