MyFavMirrors/minio
1
0
Fork 0
mirror of https://github.com/minio/minio.git synced 2025-02-03 09:55:59 -05:00
Code Issues Packages Projects Releases Wiki Activity

Add tags to auditLogLifecycle (#17081)

Browse Source
This commit is contained in:
Krishnan Parthasarathi 2023-04-26 17:49:00 -07:00 committed by GitHub
parent 31b5acc245
commit e7cac8acef
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 122 additions and 116 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

158
cmd/bucket-lifecycle.go
View File

@ -96,9 +96,8 @@ func (sys *LifecycleSys) trace(oi ObjectInfo) func(event string) {
}
type expiryTask struct {
objInfo ObjectInfo
versionExpiry bool
restoredObject bool
objInfo ObjectInfo
event lifecycle.Event
}
type expiryState struct {
@ -121,11 +120,11 @@ func (es *expiryState) close() {
}
// enqueueByDays enqueues object versions expired by days for expiry.
func (es *expiryState) enqueueByDays(oi ObjectInfo, restoredObject bool, rmVersion bool) {
func (es *expiryState) enqueueByDays(oi ObjectInfo, event lifecycle.Event) {
select {
case <-GlobalContext.Done():
es.close()
case es.byDaysCh <- expiryTask{objInfo: oi, versionExpiry: rmVersion, restoredObject: restoredObject}:
case es.byDaysCh <- expiryTask{objInfo: oi, event: event}:
default:
}
}
@ -171,9 +170,9 @@ func initBackgroundExpiry(ctx context.Context, objectAPI ObjectLayer) {
go func(t expiryTask) {
defer ewk.Give()
if t.objInfo.TransitionedObject.Status != "" {
applyExpiryOnTransitionedObject(ctx, objectAPI, t.objInfo, t.restoredObject)
applyExpiryOnTransitionedObject(ctx, objectAPI, t.objInfo, t.event)
} else {
applyExpiryOnNonTransitionedObjects(ctx, objectAPI, t.objInfo, t.versionExpiry)
applyExpiryOnNonTransitionedObjects(ctx, objectAPI, t.objInfo, t.event)
}
}(t)
}
@ -199,8 +198,8 @@ type newerNoncurrentTask struct {
}
type transitionTask struct {
tier string
objInfo ObjectInfo
event lifecycle.Event
}
type transitionState struct {
@ -218,10 +217,10 @@ type transitionState struct {
lastDayStats map[string]*lastDayTierStats
}
func (t *transitionState) queueTransitionTask(oi ObjectInfo, sc string) {
func (t *transitionState) queueTransitionTask(oi ObjectInfo, event lifecycle.Event) {
select {
case <-t.ctx.Done():
case t.transitionCh <- transitionTask{objInfo: oi, tier: sc}:
case t.transitionCh <- transitionTask{objInfo: oi, event: event}:
default:
}
}
@ -274,9 +273,9 @@ func (t *transitionState) worker(objectAPI ObjectLayer) {
return
}
atomic.AddInt32(&t.activeTasks, 1)
if err := transitionObject(t.ctx, objectAPI, task.objInfo, task.tier); err != nil {
logger.LogIf(t.ctx, fmt.Errorf("Transition failed for %s/%s version:%s with %w",
task.objInfo.Bucket, task.objInfo.Name, task.objInfo.VersionID, err))
if err := transitionObject(t.ctx, objectAPI, task.objInfo, task.event); err != nil {
logger.LogIf(t.ctx, fmt.Errorf("Transition to %s failed for %s/%s version:%s with %w",
task.event.StorageClass, task.objInfo.Bucket, task.objInfo.Name, task.objInfo.VersionID, err))
} else {
ts := tierStats{
TotalSize: uint64(task.objInfo.Size),
@ -285,7 +284,7 @@ func (t *transitionState) worker(objectAPI ObjectLayer) {
if task.objInfo.IsLatest {
ts.NumObjects = 1
}
t.addLastDayStats(task.tier, ts)
t.addLastDayStats(task.event.StorageClass, ts)
}
atomic.AddInt32(&t.activeTasks, -1)
}
@ -358,89 +357,76 @@ func validateTransitionTier(lc *lifecycle.Lifecycle) error {
// This is to be called after a successful upload of an object (version).
func enqueueTransitionImmediate(obj ObjectInfo) {
if lc, err := globalLifecycleSys.Get(obj.Bucket); err == nil {
event := lc.Eval(obj.ToLifecycleOpts())
switch event.Action {
switch event := lc.Eval(obj.ToLifecycleOpts()); event.Action {
case lifecycle.TransitionAction, lifecycle.TransitionVersionAction:
globalTransitionState.queueTransitionTask(obj, event.StorageClass)
globalTransitionState.queueTransitionTask(obj, event)
}
}
}
// expireAction represents different actions to be performed on expiry of a
// restored/transitioned object
type expireAction int
const (
// ignore the zero value
_ expireAction = iota
// expireObj indicates expiry of 'regular' transitioned objects.
expireObj
// expireRestoredObj indicates expiry of restored objects.
expireRestoredObj
)
// expireTransitionedObject handles expiry of transitioned/restored objects
// (versions) in one of the following situations:
//
// 1. when a restored (via PostRestoreObject API) object expires.
// 2. when a transitioned object expires (based on an ILM rule).
func expireTransitionedObject(ctx context.Context, objectAPI ObjectLayer, oi *ObjectInfo, lcOpts lifecycle.ObjectOpts, action expireAction) error {
func expireTransitionedObject(ctx context.Context, objectAPI ObjectLayer, oi *ObjectInfo, lcOpts lifecycle.ObjectOpts, lcEvent lifecycle.Event) error {
traceFn := globalLifecycleSys.trace(*oi)
var opts ObjectOptions
opts.Versioned = globalBucketVersioningSys.PrefixEnabled(oi.Bucket, oi.Name)
opts.VersionID = lcOpts.VersionID
opts.Expiration = ExpirationOptions{Expire: true}
switch action {
case expireObj:
// When an object is past expiry or when a transitioned object is being
// deleted, 'mark' the data in the remote tier for delete.
entry := jentry{
ObjName: oi.TransitionedObject.Name,
VersionID: oi.TransitionedObject.VersionID,
TierName: oi.TransitionedObject.Tier,
}
if err := globalTierJournal.AddEntry(entry); err != nil {
logger.LogIf(ctx, err)
return err
}
// Delete metadata on source, now that data in remote tier has been
// marked for deletion.
if _, err := objectAPI.DeleteObject(ctx, oi.Bucket, oi.Name, opts); err != nil {
logger.LogIf(ctx, err)
return err
}
// Send audit for the lifecycle delete operation
defer auditLogLifecycle(ctx, *oi, ILMExpiry, traceFn)
eventName := event.ObjectRemovedDelete
if lcOpts.DeleteMarker {
eventName = event.ObjectRemovedDeleteMarkerCreated
}
objInfo := ObjectInfo{
Name: oi.Name,
VersionID: lcOpts.VersionID,
DeleteMarker: lcOpts.DeleteMarker,
}
// Notify object deleted event.
sendEvent(eventArgs{
EventName: eventName,
BucketName: oi.Bucket,
Object: objInfo,
UserAgent: "Internal: [ILM-Expiry]",
Host: globalLocalNodeName,
})
case expireRestoredObj:
tags := auditLifecycleTags(lcEvent)
if lcEvent.Action.DeleteRestored() {
// delete locally restored copy of object or object version
// from the source, while leaving metadata behind. The data on
// transitioned tier lies untouched and still accessible
opts.Transition.ExpireRestored = true
_, err := objectAPI.DeleteObject(ctx, oi.Bucket, oi.Name, opts)
if err == nil {
// TODO consider including expiry of restored object to events we
// notify.
auditLogLifecycle(ctx, *oi, ILMExpiry, tags, traceFn)
}
return err
default:
return fmt.Errorf("Unknown expire action %v", action)
}
// When an object is past expiry or when a transitioned object is being
// deleted, 'mark' the data in the remote tier for delete.
entry := jentry{
ObjName: oi.TransitionedObject.Name,
VersionID: oi.TransitionedObject.VersionID,
TierName: oi.TransitionedObject.Tier,
}
if err := globalTierJournal.AddEntry(entry); err != nil {
logger.LogIf(ctx, err)
return err
}
// Delete metadata on source, now that data in remote tier has been
// marked for deletion.
if _, err := objectAPI.DeleteObject(ctx, oi.Bucket, oi.Name, opts); err != nil {
logger.LogIf(ctx, err)
return err
}
// Send audit for the lifecycle delete operation
defer auditLogLifecycle(ctx, *oi, ILMExpiry, tags, traceFn)
eventName := event.ObjectRemovedDelete
if lcOpts.DeleteMarker {
eventName = event.ObjectRemovedDeleteMarkerCreated
}
objInfo := ObjectInfo{
Name: oi.Name,
VersionID: lcOpts.VersionID,
DeleteMarker: lcOpts.DeleteMarker,
}
// Notify object deleted event.
sendEvent(eventArgs{
EventName: eventName,
BucketName: oi.Bucket,
Object: objInfo,
UserAgent: "Internal: [ILM-Expiry]",
Host: globalLocalNodeName,
})
return nil
}
@ -460,13 +446,14 @@ func genTransitionObjName(bucket string) (string, error) {
// storage specified by the transition ARN, the metadata is left behind on source cluster and original content
// is moved to the transition tier. Note that in the case of encrypted objects, entire encrypted stream is moved
// to the transition tier without decrypting or re-encrypting.
func transitionObject(ctx context.Context, objectAPI ObjectLayer, oi ObjectInfo, tier string) error {
func transitionObject(ctx context.Context, objectAPI ObjectLayer, oi ObjectInfo, lcEvent lifecycle.Event) error {
opts := ObjectOptions{
Transition: TransitionOptions{
Status: lifecycle.TransitionPending,
Tier: tier,
Tier: lcEvent.StorageClass,
ETag: oi.ETag,
},
LifecycleEvent: lcEvent,
VersionID: oi.VersionID,
Versioned: globalBucketVersioningSys.PrefixEnabled(oi.Bucket, oi.Name),
VersionSuspended: globalBucketVersioningSys.PrefixSuspended(oi.Bucket, oi.Name),
@ -878,3 +865,20 @@ func (oi ObjectInfo) ToLifecycleOpts() lifecycle.ObjectOpts {
TransitionStatus: oi.TransitionedObject.Status,
}
}
func auditLifecycleTags(event lifecycle.Event) map[string]interface{} {
const (
ilmAction = "ilm-action"
ilmDue = "ilm-due"
ilmRuleID = "ilm-rule-id"
ilmTier = "ilm-tier"
)
tags := make(map[string]interface{}, 4)
tags[ilmAction] = event.Action.String()
tags[ilmDue] = event.Due
tags[ilmRuleID] = event.RuleID
if event.StorageClass != "" {
tags[ilmTier] = event.StorageClass
}
return tags
}

45
cmd/data-scanner.go
View File

@ -945,9 +945,9 @@ func (i *scannerItem) applyLifecycle(ctx context.Context, o ObjectLayer, oi Obje
switch lcEvt.Action {
case lifecycle.DeleteAction, lifecycle.DeleteVersionAction, lifecycle.DeleteRestoredAction, lifecycle.DeleteRestoredVersionAction:
return applyLifecycleAction(lcEvt.Action, oi, ""), 0
return applyLifecycleAction(lcEvt, oi), 0
case lifecycle.TransitionAction, lifecycle.TransitionVersionAction:
return applyLifecycleAction(lcEvt.Action, oi, lcEvt.StorageClass), size
return applyLifecycleAction(lcEvt, oi), size
default:
// No action.
return false, size
@ -983,7 +983,7 @@ func (i *scannerItem) applyTierObjSweep(ctx context.Context, o ObjectLayer, oi O
InclFreeVersions: true,
})
if err == nil {
auditLogLifecycle(ctx, oi, ILMFreeVersionDelete, traceFn)
auditLogLifecycle(ctx, oi, ILMFreeVersionDelete, nil, traceFn)
}
if ignoreNotFoundErr(err) != nil {
logger.LogIf(ctx, err)
@ -1153,20 +1153,16 @@ func evalActionFromLifecycle(ctx context.Context, lc lifecycle.Lifecycle, lr loc
return event
}
func applyTransitionRule(obj ObjectInfo, storageClass string) bool {
func applyTransitionRule(event lifecycle.Event, obj ObjectInfo) bool {
if obj.DeleteMarker {
return false
}
globalTransitionState.queueTransitionTask(obj, storageClass)
globalTransitionState.queueTransitionTask(obj, event)
return true
}
func applyExpiryOnTransitionedObject(ctx context.Context, objLayer ObjectLayer, obj ObjectInfo, restoredObject bool) bool {
action := expireObj
if restoredObject {
action = expireRestoredObj
}
if err := expireTransitionedObject(ctx, objLayer, &obj, obj.ToLifecycleOpts(), action); err != nil {
func applyExpiryOnTransitionedObject(ctx context.Context, objLayer ObjectLayer, obj ObjectInfo, lcEvent lifecycle.Event) bool {
if err := expireTransitionedObject(ctx, objLayer, &obj, obj.ToLifecycleOpts(), lcEvent); err != nil {
if isErrObjectNotFound(err) || isErrVersionNotFound(err) {
return false
}
@ -1177,13 +1173,13 @@ func applyExpiryOnTransitionedObject(ctx context.Context, objLayer ObjectLayer,
return true
}
func applyExpiryOnNonTransitionedObjects(ctx context.Context, objLayer ObjectLayer, obj ObjectInfo, applyOnVersion bool) bool {
func applyExpiryOnNonTransitionedObjects(ctx context.Context, objLayer ObjectLayer, obj ObjectInfo, lcEvent lifecycle.Event) bool {
traceFn := globalLifecycleSys.trace(obj)
opts := ObjectOptions{
Expiration: ExpirationOptions{Expire: true},
}
if applyOnVersion {
if lcEvent.Action.DeleteVersioned() {
opts.VersionID = obj.VersionID
}
if opts.VersionID == "" {
@ -1201,8 +1197,9 @@ func applyExpiryOnNonTransitionedObjects(ctx context.Context, objLayer ObjectLay
return false
}
tags := auditLifecycleTags(lcEvent)
// Send audit for the lifecycle delete operation
auditLogLifecycle(ctx, obj, ILMExpiry, traceFn)
auditLogLifecycle(ctx, obj, ILMExpiry, tags, traceFn)
eventName := event.ObjectRemovedDelete
if obj.DeleteMarker {
@ -1222,20 +1219,19 @@ func applyExpiryOnNonTransitionedObjects(ctx context.Context, objLayer ObjectLay
}
// Apply object, object version, restored object or restored object version action on the given object
func applyExpiryRule(obj ObjectInfo, restoredObject, applyOnVersion bool) bool {
globalExpiryState.enqueueByDays(obj, restoredObject, applyOnVersion)
func applyExpiryRule(event lifecycle.Event, obj ObjectInfo) bool {
globalExpiryState.enqueueByDays(obj, event)
return true
}
// Perform actions (removal or transitioning of objects), return true the action is successfully performed
func applyLifecycleAction(action lifecycle.Action, obj ObjectInfo, storageClass string) (success bool) {
switch action {
case lifecycle.DeleteVersionAction, lifecycle.DeleteAction:
success = applyExpiryRule(obj, false, action == lifecycle.DeleteVersionAction)
case lifecycle.DeleteRestoredAction, lifecycle.DeleteRestoredVersionAction:
success = applyExpiryRule(obj, true, action == lifecycle.DeleteRestoredVersionAction)
func applyLifecycleAction(event lifecycle.Event, obj ObjectInfo) (success bool) {
switch action := event.Action; action {
case lifecycle.DeleteVersionAction, lifecycle.DeleteAction,
lifecycle.DeleteRestoredAction, lifecycle.DeleteRestoredVersionAction:
success = applyExpiryRule(event, obj)
case lifecycle.TransitionAction, lifecycle.TransitionVersionAction:
success = applyTransitionRule(obj, storageClass)
success = applyTransitionRule(event, obj)
}
return
}
@ -1445,7 +1441,7 @@ const (
ILMTransition = " ilm:transition"
)
func auditLogLifecycle(ctx context.Context, oi ObjectInfo, event string, traceFn func(event string)) {
func auditLogLifecycle(ctx context.Context, oi ObjectInfo, event string, tags map[string]interface{}, traceFn func(event string)) {
var apiName string
switch event {
case ILMExpiry:
@ -1461,6 +1457,7 @@ func auditLogLifecycle(ctx context.Context, oi ObjectInfo, event string, traceFn
Bucket: oi.Bucket,
Object: oi.Name,
VersionID: oi.VersionID,
Tags: tags,
})
traceFn(event)
}

3
cmd/erasure-object.go
View File

@ -2004,7 +2004,8 @@ func (er erasureObjects) TransitionObject(ctx context.Context, bucket, object st
UserAgent: "Internal: [ILM-Transition]",
Host: globalLocalNodeName,
})
auditLogLifecycle(ctx, objInfo, ILMTransition, traceFn)
tags := auditLifecycleTags(opts.LifecycleEvent)
auditLogLifecycle(ctx, objInfo, ILMTransition, tags, traceFn)
return err
}

4
cmd/erasure-server-pool-decom.go
View File

@ -738,10 +738,10 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
evt := evalActionFromLifecycle(ctx, *lc, lr, objInfo)
switch {
case evt.Action.DeleteRestored(): // if restored copy has expired,delete it synchronously
applyExpiryOnTransitionedObject(ctx, z, objInfo, evt.Action.DeleteRestored())
applyExpiryOnTransitionedObject(ctx, z, objInfo, evt)
return false
case evt.Action.Delete():
globalExpiryState.enqueueByDays(objInfo, evt.Action.DeleteRestored(), evt.Action.DeleteVersioned())
globalExpiryState.enqueueByDays(objInfo, evt)
return true
default:
return false

2
cmd/erasure-server-pool-rebalance.go
View File

@ -465,7 +465,7 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
evt := evalActionFromLifecycle(ctx, *lc, lr, objInfo)
if evt.Action.Delete() {
globalExpiryState.enqueueByDays(objInfo, evt.Action.DeleteRestored(), evt.Action.DeleteVersioned())
globalExpiryState.enqueueByDays(objInfo, evt)
return true
}

2
cmd/erasure-server-pool.go
View File

@ -1320,7 +1320,7 @@ func (z *erasureServerPools) ListObjects(ctx context.Context, bucket, prefix, ma
if opts.Lifecycle != nil {
evt := evalActionFromLifecycle(ctx, *opts.Lifecycle, opts.Retention, objInfo)
if evt.Action.Delete() {
globalExpiryState.enqueueByDays(objInfo, evt.Action.DeleteRestored(), evt.Action.DeleteVersioned())
globalExpiryState.enqueueByDays(objInfo, evt)
if !evt.Action.DeleteRestored() {
// Skip entry if ILM action was DeleteVersionAction or DeleteAction
return loi, nil

2
cmd/metacache-server-pool.go
View File

@ -395,7 +395,7 @@ func applyBucketActions(ctx context.Context, o listPathOptions, in <-chan metaCa
if o.Lifecycle != nil {
evt := evalActionFromLifecycle(ctx, *o.Lifecycle, o.Retention, objInfo)
if evt.Action.Delete() {
globalExpiryState.enqueueByDays(objInfo, evt.Action.DeleteRestored(), evt.Action.DeleteVersioned())
globalExpiryState.enqueueByDays(objInfo, evt)
if !evt.Action.DeleteRestored() {
continue
} // queue version for replication upon expired restored copies if needed.

2
cmd/object-api-interface.go
View File

@ -28,6 +28,7 @@ import (
"github.com/minio/minio-go/v7/pkg/tags"
"github.com/minio/minio/internal/hash"
"github.com/minio/minio/internal/bucket/lifecycle"
"github.com/minio/minio/internal/bucket/replication"
xioutil "github.com/minio/minio/internal/ioutil"
)
@ -60,6 +61,7 @@ type ObjectOptions struct {
DeleteReplication ReplicationState // Represents internal replication state needed for Delete replication
Transition TransitionOptions
Expiration ExpirationOptions
LifecycleEvent lifecycle.Event
WantChecksum *hash.Checksum // x-amz-checksum-XXX checksum sent to PutObject/ CompleteMultipartUpload.

4
cmd/object-handlers-common.go
View File

@ -373,11 +373,13 @@ func deleteObjectVersions(ctx context.Context, o ObjectLayer, bucket string, toD
VersionID: dobj.VersionID,
}
traceFn := globalLifecycleSys.trace(oi)
tags := make(map[string]interface{}, 1)
tags["newer-noncurrent-versions"] = true
// Send audit for the lifecycle delete operation
auditLogLifecycle(
ctx,
oi,
ILMExpiry, traceFn)
ILMExpiry, tags, traceFn)
sendEvent(eventArgs{
EventName: event.ObjectRemovedDelete,

16
cmd/object-handlers.go
View File

@ -476,11 +476,11 @@ func (api objectAPIHandlers) getObjectHandler(ctx context.Context, objectAPI Obj
// Automatically remove the object/version if an expiry lifecycle rule can be applied
if lc, err := globalLifecycleSys.Get(bucket); err == nil {
rcfg, _ := globalBucketObjectLockSys.Get(bucket)
act := evalActionFromLifecycle(ctx, *lc, rcfg, objInfo).Action
if act.Delete() {
event := evalActionFromLifecycle(ctx, *lc, rcfg, objInfo)
if event.Action.Delete() {
// apply whatever the expiry rule is.
applyExpiryRule(objInfo, act.DeleteRestored(), act.DeleteVersioned())
if !act.DeleteRestored() {
applyExpiryRule(event, objInfo)
if !event.Action.DeleteRestored() {
// If the ILM action is not on restored object return error.
writeErrorResponseHeadersOnly(w, errorCodes.ToAPIErr(ErrNoSuchKey))
return
@ -729,11 +729,11 @@ func (api objectAPIHandlers) headObjectHandler(ctx context.Context, objectAPI Ob
// Automatically remove the object/version if an expiry lifecycle rule can be applied
if lc, err := globalLifecycleSys.Get(bucket); err == nil {
rcfg, _ := globalBucketObjectLockSys.Get(bucket)
act := evalActionFromLifecycle(ctx, *lc, rcfg, objInfo).Action
if act.Delete() {
event := evalActionFromLifecycle(ctx, *lc, rcfg, objInfo)
if event.Action.Delete() {
// apply whatever the expiry rule is.
applyExpiryRule(objInfo, act.DeleteRestored(), act.DeleteVersioned())
if !act.DeleteRestored() {
applyExpiryRule(event, objInfo)
if !event.Action.DeleteRestored() {
// If the ILM action is not on restored object return error.
writeErrorResponseHeadersOnly(w, errorCodes.ToAPIErr(ErrNoSuchKey))
return