mirror of
https://github.com/pgsty/minio.git
synced 2026-07-23 06:00:24 +03:00
reduce extra getObjectInfo() calls during ILM transition (#13091)
* reduce extra getObjectInfo() calls during ILM transition This PR also changes expiration logic to be non-blocking, scanner is now free from additional costs incurred due to slower object layer calls and hitting the drives. * move verifying expiration inside locks
This commit is contained in:
+56
-10
@@ -1191,9 +1191,40 @@ func (er erasureObjects) DeleteObject(ctx context.Context, bucket, object string
|
||||
return ObjectInfo{}, toObjectErr(er.deletePrefix(ctx, bucket, object), bucket, object)
|
||||
}
|
||||
|
||||
var lc *lifecycle.Lifecycle
|
||||
if opts.Expiration.Expire {
|
||||
// Check if the current bucket has a configured lifecycle policy
|
||||
lc, _ = globalLifecycleSys.Get(bucket)
|
||||
}
|
||||
|
||||
// expiration attempted on a bucket with no lifecycle
|
||||
// rules shall be rejected.
|
||||
if lc == nil && opts.Expiration.Expire {
|
||||
if opts.VersionID != "" {
|
||||
return objInfo, VersionNotFound{
|
||||
Bucket: bucket,
|
||||
Object: object,
|
||||
VersionID: opts.VersionID,
|
||||
}
|
||||
}
|
||||
return objInfo, ObjectNotFound{
|
||||
Bucket: bucket,
|
||||
Object: object,
|
||||
}
|
||||
}
|
||||
|
||||
// Acquire a write lock before deleting the object.
|
||||
lk := er.NewNSLock(bucket, object)
|
||||
lkctx, err := lk.GetLock(ctx, globalDeleteOperationTimeout)
|
||||
if err != nil {
|
||||
return ObjectInfo{}, err
|
||||
}
|
||||
ctx = lkctx.Context()
|
||||
defer lk.Unlock(lkctx.Cancel)
|
||||
|
||||
versionFound := true
|
||||
objInfo = ObjectInfo{VersionID: opts.VersionID} // version id needed in Delete API response.
|
||||
goi, gerr := er.GetObjectInfo(ctx, bucket, object, opts)
|
||||
goi, gerr := er.getObjectInfo(ctx, bucket, object, opts)
|
||||
if gerr != nil && goi.Name == "" {
|
||||
switch gerr.(type) {
|
||||
case InsufficientReadQuorum:
|
||||
@@ -1207,16 +1238,31 @@ func (er erasureObjects) DeleteObject(ctx context.Context, bucket, object string
|
||||
}
|
||||
}
|
||||
|
||||
defer NSUpdated(bucket, object)
|
||||
|
||||
// Acquire a write lock before deleting the object.
|
||||
lk := er.NewNSLock(bucket, object)
|
||||
lkctx, err := lk.GetLock(ctx, globalDeleteOperationTimeout)
|
||||
if err != nil {
|
||||
return ObjectInfo{}, err
|
||||
if opts.Expiration.Expire {
|
||||
action := evalActionFromLifecycle(ctx, *lc, goi, false)
|
||||
var isErr bool
|
||||
switch action {
|
||||
case lifecycle.NoneAction:
|
||||
isErr = true
|
||||
case lifecycle.TransitionAction, lifecycle.TransitionVersionAction:
|
||||
isErr = true
|
||||
}
|
||||
if isErr {
|
||||
if goi.VersionID != "" {
|
||||
return goi, VersionNotFound{
|
||||
Bucket: bucket,
|
||||
Object: object,
|
||||
VersionID: goi.VersionID,
|
||||
}
|
||||
}
|
||||
return goi, ObjectNotFound{
|
||||
Bucket: bucket,
|
||||
Object: object,
|
||||
}
|
||||
}
|
||||
}
|
||||
ctx = lkctx.Context()
|
||||
defer lk.Unlock(lkctx.Cancel)
|
||||
|
||||
defer NSUpdated(bucket, object)
|
||||
|
||||
storageDisks := er.getDisks()
|
||||
writeQuorum := len(storageDisks)/2 + 1
|
||||
|
||||
Reference in New Issue
Block a user