gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(objectnode): lifecycle skip locked object

Browse Source 
Signed-off-by: tangdeyi <tangdeyi@oppo.com>
This commit is contained in:
tangdeyi 2023-10-13 15:15:08 +08:00 committed by tangdeyi
parent a5dd54d60f
commit fcdeaebb69
3 changed files with 28 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
objectnode/api_handler_object.go
View File

@ -654,7 +654,7 @@ func (o *ObjectNode) deleteObjectsHandler(w http.ResponseWriter, r *http.Request
if err1 := vol.DeletePath(object.Key); err1 != nil {
log.LogErrorf("deleteObjectsHandler: delete object failed: requestID(%v) volume(%v) path(%v) err(%v)",
GetRequestID(r), vol.Name(), object.Key, err1)
if err1 != AccessDenied {
if !strings.Contains(err1.Error(), AccessDenied.ErrorMessage) {
deletedErrors = append(deletedErrors, Error{Key: object.Key, Code: "InternalError", Message: err1.Error()})
} else {
deletedErrors = append(deletedErrors, Error{Key: object.Key, Code: "AccessDenied", Message: err1.Error()})

6
objectnode/fs_volume.go
View File

@ -899,12 +899,6 @@ func (v *Volume) DeletePath(path string) (err error) {
log.LogErrorf("DeletePath: load volume objetLock: volume(%v) err(%v)", v.name, err)
return
}
if objetLock != nil {
err = isObjectLocked(v, ino, name, path)
if err != nil {
return
}
}
log.LogInfof("DeletePath: delete: volume(%v) path(%v) inode(%v)", v.name, path, ino)
// delete dentry with condition when objectlock is open

28
sdk/meta/api.go
View File

@ -15,6 +15,7 @@
package meta
import (
"errors"
"fmt"
syslog "log"
"math"
@ -810,12 +811,37 @@ func (mw *MetaWrapper) Delete_ll_EX(parentID uint64, name string, isDir bool, ve
return info, nil
}
func isObjectLocked(mw *MetaWrapper, inode uint64, name string) error {
xattrInfo, err := mw.XAttrGet_ll(inode, "oss:lock")
if err != nil {
log.LogErrorf("isObjectLocked: check ObjectLock err(%v) name(%v)", err, name)
return err
}
retainUntilDate := xattrInfo.Get("oss:lock")
if len(retainUntilDate) > 0 {
retainUntilDateInt64, err := strconv.ParseInt(string(retainUntilDate), 10, 64)
if err != nil {
return err
}
if retainUntilDateInt64 > time.Now().UnixNano() {
log.LogWarnf("isObjectLocked: object is locked, retainUntilDate(%v) name(%v)", retainUntilDateInt64, name)
return errors.New("Access Denied")
}
}
return nil
}
func (mw *MetaWrapper) deletewithcond_ll(parentID, cond uint64, name string, isDir bool) (*proto.InodeInfo, error) {
err := isObjectLocked(mw, cond, name)
if err != nil {
return nil, err
}
var (
status int
inode uint64
mode uint32
err error
info *proto.InodeInfo
mp *MetaPartition
)