MyFavMirrors
/
minio
mirror of https://github.com/minio/minio.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add lifecycle event source to audit log tags (#17248)

This commit is contained in:
Krishnan Parthasarathi 2023-05-22 15:28:56 -07:00 committed by GitHub
parent 55a3310446
commit 3e128c116e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 176 additions and 86 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

88
cmd/bucket-lifecycle-audit.go Normal file
View File

@ -0,0 +1,88 @@
// Copyright (c) 2023 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package cmd
import "github.com/minio/minio/internal/bucket/lifecycle"
//go:generate stringer -type lcEventSrc -trimprefix lcEventSrc_ $GOFILE
type lcEventSrc uint8
//revive:disable:var-naming Underscores is used here to indicate where common prefix ends and the enumeration name begins
const (
lcEventSrc_None lcEventSrc = iota
lcEventSrc_Scanner
lcEventSrc_Decom
lcEventSrc_Rebal
lcEventSrc_s3HeadObject
lcEventSrc_s3GetObject
lcEventSrc_s3ListObjects
lcEventSrc_s3PutObject
lcEventSrc_s3CopyObject
lcEventSrc_s3CompleteMultipartUpload
)
//revive:enable:var-naming
type lcAuditEvent struct {
lifecycle.Event
source lcEventSrc
}
func (lae lcAuditEvent) Tags() map[string]interface{} {
event := lae.Event
src := lae.source
const (
ilmSrc = "ilm-src"
ilmAction = "ilm-action"
ilmDue = "ilm-due"
ilmRuleID = "ilm-rule-id"
ilmTier = "ilm-tier"
ilmNewerNoncurrentVersions = "ilm-newer-noncurrent-versions"
ilmNoncurrentDays = "ilm-noncurrent-days"
)
tags := make(map[string]interface{}, 5)
if src > lcEventSrc_None {
tags[ilmSrc] = src.String()
}
tags[ilmAction] = event.Action.String()
tags[ilmRuleID] = event.RuleID
if !event.Due.IsZero() {
tags[ilmDue] = event.Due
}
// rule with Transition/NoncurrentVersionTransition in effect
if event.StorageClass != "" {
tags[ilmTier] = event.StorageClass
}
// rule with NewernoncurrentVersions in effect
if event.NewerNoncurrentVersions > 0 {
tags[ilmNewerNoncurrentVersions] = event.NewerNoncurrentVersions
}
if event.NoncurrentDays > 0 {
tags[ilmNoncurrentDays] = event.NoncurrentDays
}
return tags
}
func newLifecycleAuditEvent(src lcEventSrc, event lifecycle.Event) lcAuditEvent {
return lcAuditEvent{
Event: event,
source: src,
}
}

62
cmd/bucket-lifecycle.go
View File

@ -98,6 +98,7 @@ func (sys *LifecycleSys) trace(oi ObjectInfo) func(event string) {
type expiryTask struct {
objInfo ObjectInfo
event lifecycle.Event
src lcEventSrc
}
type expiryState struct {
@ -120,11 +121,11 @@ func (es *expiryState) close() {
}
// enqueueByDays enqueues object versions expired by days for expiry.
func (es *expiryState) enqueueByDays(oi ObjectInfo, event lifecycle.Event) {
func (es *expiryState) enqueueByDays(oi ObjectInfo, event lifecycle.Event, src lcEventSrc) {
select {
case <-GlobalContext.Done():
es.close()
case es.byDaysCh <- expiryTask{objInfo: oi, event: event}:
case es.byDaysCh <- expiryTask{objInfo: oi, event: event, src: src}:
default:
}
}
@ -172,9 +173,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.event)
applyExpiryOnTransitionedObject(ctx, objectAPI, t.objInfo, t.event, t.src)
} else {
applyExpiryOnNonTransitionedObjects(ctx, objectAPI, t.objInfo, t.event)
applyExpiryOnNonTransitionedObjects(ctx, objectAPI, t.objInfo, t.event, t.src)
}
}(t)
}
@ -202,6 +203,7 @@ type newerNoncurrentTask struct {
type transitionTask struct {
objInfo ObjectInfo
src lcEventSrc
event lifecycle.Event
}
@ -220,10 +222,10 @@ type transitionState struct {
lastDayStats map[string]*lastDayTierStats
}
func (t *transitionState) queueTransitionTask(oi ObjectInfo, event lifecycle.Event) {
func (t *transitionState) queueTransitionTask(oi ObjectInfo, event lifecycle.Event, src lcEventSrc) {
select {
case <-t.ctx.Done():
case t.transitionCh <- transitionTask{objInfo: oi, event: event}:
case t.transitionCh <- transitionTask{objInfo: oi, event: event, src: src}:
default:
}
}
@ -276,7 +278,7 @@ func (t *transitionState) worker(objectAPI ObjectLayer) {
return
}
atomic.AddInt32(&t.activeTasks, 1)
if err := transitionObject(t.ctx, objectAPI, task.objInfo, task.event); err != nil {
if err := transitionObject(t.ctx, objectAPI, task.objInfo, newLifecycleAuditEvent(task.src, 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 {
@ -358,11 +360,11 @@ func validateTransitionTier(lc *lifecycle.Lifecycle) error {
// enqueueTransitionImmediate enqueues obj for transition if eligible.
// This is to be called after a successful upload of an object (version).
func enqueueTransitionImmediate(obj ObjectInfo) {
func enqueueTransitionImmediate(obj ObjectInfo, src lcEventSrc) {
if lc, err := globalLifecycleSys.Get(obj.Bucket); err == nil {
switch event := lc.Eval(obj.ToLifecycleOpts()); event.Action {
case lifecycle.TransitionAction, lifecycle.TransitionVersionAction:
globalTransitionState.queueTransitionTask(obj, event)
globalTransitionState.queueTransitionTask(obj, event, src)
}
}
}
@ -372,13 +374,13 @@ func enqueueTransitionImmediate(obj ObjectInfo) {
//
// 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, lcEvent lifecycle.Event) error {
func expireTransitionedObject(ctx context.Context, objectAPI ObjectLayer, oi *ObjectInfo, lcOpts lifecycle.ObjectOpts, lcEvent lifecycle.Event, src lcEventSrc) 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}
tags := auditLifecycleTags(lcEvent)
tags := newLifecycleAuditEvent(src, lcEvent).Tags()
if lcEvent.Action.DeleteRestored() {
// delete locally restored copy of object or object version
// from the source, while leaving metadata behind. The data on
@ -449,14 +451,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, lcEvent lifecycle.Event) error {
func transitionObject(ctx context.Context, objectAPI ObjectLayer, oi ObjectInfo, lae lcAuditEvent) error {
opts := ObjectOptions{
Transition: TransitionOptions{
Status: lifecycle.TransitionPending,
Tier: lcEvent.StorageClass,
Tier: lae.StorageClass,
ETag: oi.ETag,
},
LifecycleEvent: lcEvent,
LifecycleAuditEvent: lae,
VersionID: oi.VersionID,
Versioned: globalBucketVersioningSys.PrefixEnabled(oi.Bucket, oi.Name),
VersionSuspended: globalBucketVersioningSys.PrefixSuspended(oi.Bucket, oi.Name),
@ -868,35 +870,3 @@ 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"
ilmNewerNoncurrentVersions = "ilm-newer-noncurrent-versions"
ilmNoncurrentDays = "ilm-noncurrent-days"
)
tags := make(map[string]interface{}, 4)
tags[ilmAction] = event.Action.String()
tags[ilmRuleID] = event.RuleID
if !event.Due.IsZero() {
tags[ilmDue] = event.Due
}
// rule with Transition/NoncurrentVersionTransition in effect
if event.StorageClass != "" {
tags[ilmTier] = event.StorageClass
}
// rule with NewernoncurrentVersions in effect
if event.NewerNoncurrentVersions > 0 {
tags[ilmNewerNoncurrentVersions] = event.NewerNoncurrentVersions
}
if event.NoncurrentDays > 0 {
tags[ilmNoncurrentDays] = event.NoncurrentDays
}
return tags
}

26
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, oi), 0
return applyLifecycleAction(lcEvt, lcEventSrc_Scanner, oi), 0
case lifecycle.TransitionAction, lifecycle.TransitionVersionAction:
return applyLifecycleAction(lcEvt, oi), size
return applyLifecycleAction(lcEvt, lcEventSrc_Scanner, oi), size
default:
// No action.
return false, size
@ -1154,16 +1154,16 @@ func evalActionFromLifecycle(ctx context.Context, lc lifecycle.Lifecycle, lr loc
return event
}
func applyTransitionRule(event lifecycle.Event, obj ObjectInfo) bool {
func applyTransitionRule(event lifecycle.Event, src lcEventSrc, obj ObjectInfo) bool {
if obj.DeleteMarker {
return false
}
globalTransitionState.queueTransitionTask(obj, event)
globalTransitionState.queueTransitionTask(obj, event, src)
return true
}
func applyExpiryOnTransitionedObject(ctx context.Context, objLayer ObjectLayer, obj ObjectInfo, lcEvent lifecycle.Event) bool {
if err := expireTransitionedObject(ctx, objLayer, &obj, obj.ToLifecycleOpts(), lcEvent); err != nil {
func applyExpiryOnTransitionedObject(ctx context.Context, objLayer ObjectLayer, obj ObjectInfo, lcEvent lifecycle.Event, src lcEventSrc) bool {
if err := expireTransitionedObject(ctx, objLayer, &obj, obj.ToLifecycleOpts(), lcEvent, src); err != nil {
if isErrObjectNotFound(err) || isErrVersionNotFound(err) {
return false
}
@ -1174,7 +1174,7 @@ func applyExpiryOnTransitionedObject(ctx context.Context, objLayer ObjectLayer,
return true
}
func applyExpiryOnNonTransitionedObjects(ctx context.Context, objLayer ObjectLayer, obj ObjectInfo, lcEvent lifecycle.Event) bool {
func applyExpiryOnNonTransitionedObjects(ctx context.Context, objLayer ObjectLayer, obj ObjectInfo, lcEvent lifecycle.Event, src lcEventSrc) bool {
traceFn := globalLifecycleSys.trace(obj)
opts := ObjectOptions{
Expiration: ExpirationOptions{Expire: true},
@ -1198,7 +1198,7 @@ func applyExpiryOnNonTransitionedObjects(ctx context.Context, objLayer ObjectLay
return false
}
tags := auditLifecycleTags(lcEvent)
tags := newLifecycleAuditEvent(src, lcEvent).Tags()
// Send audit for the lifecycle delete operation
auditLogLifecycle(ctx, obj, ILMExpiry, tags, traceFn)
@ -1220,19 +1220,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(event lifecycle.Event, obj ObjectInfo) bool {
globalExpiryState.enqueueByDays(obj, event)
func applyExpiryRule(event lifecycle.Event, src lcEventSrc, obj ObjectInfo) bool {
globalExpiryState.enqueueByDays(obj, event, src)
return true
}
// Perform actions (removal or transitioning of objects), return true the action is successfully performed
func applyLifecycleAction(event lifecycle.Event, obj ObjectInfo) (success bool) {
func applyLifecycleAction(event lifecycle.Event, src lcEventSrc, obj ObjectInfo) (success bool) {
switch action := event.Action; action {
case lifecycle.DeleteVersionAction, lifecycle.DeleteAction,
lifecycle.DeleteRestoredAction, lifecycle.DeleteRestoredVersionAction:
success = applyExpiryRule(event, obj)
success = applyExpiryRule(event, src, obj)
case lifecycle.TransitionAction, lifecycle.TransitionVersionAction:
success = applyTransitionRule(event, obj)
success = applyTransitionRule(event, src, obj)
}
return
}

2
cmd/erasure-object.go
View File

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

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

@ -694,10 +694,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)
applyExpiryOnTransitionedObject(ctx, z, objInfo, evt, lcEventSrc_Decom)
return false
case evt.Action.Delete():
globalExpiryState.enqueueByDays(objInfo, evt)
globalExpiryState.enqueueByDays(objInfo, evt, lcEventSrc_Decom)
return true
default:
return false
@ -974,10 +974,10 @@ func (z *erasureServerPools) checkAfterDecom(ctx context.Context, idx int) error
evt := evalActionFromLifecycle(ctx, *lc, lr, objInfo)
switch {
case evt.Action.DeleteRestored(): // if restored copy has expired,delete it synchronously
applyExpiryOnTransitionedObject(ctx, z, objInfo, evt)
applyExpiryOnTransitionedObject(ctx, z, objInfo, evt, lcEventSrc_Decom)
return false
case evt.Action.Delete():
globalExpiryState.enqueueByDays(objInfo, evt)
globalExpiryState.enqueueByDays(objInfo, evt, lcEventSrc_Decom)
return true
default:
return false

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

@ -466,7 +466,7 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
evt := evalActionFromLifecycle(ctx, *lc, lr, objInfo)
if evt.Action.Delete() {
globalExpiryState.enqueueByDays(objInfo, evt)
globalExpiryState.enqueueByDays(objInfo, evt, lcEventSrc_Rebal)
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)
globalExpiryState.enqueueByDays(objInfo, evt, lcEventSrc_s3ListObjects)
if !evt.Action.DeleteRestored() {
// Skip entry if ILM action was DeleteVersionAction or DeleteAction
return loi, nil

32
cmd/lceventsrc_string.go Normal file
View File

@ -0,0 +1,32 @@
// Code generated by "stringer -type lcEventSrc -trimprefix lcEventSrc_ bucket-lifecycle-audit.go"; DO NOT EDIT.
package cmd
import "strconv"
func _() {
// An "invalid array index" compiler error signifies that the constant values have changed.
// Re-run the stringer command to generate them again.
var x [1]struct{}
_ = x[lcEventSrc_None-0]
_ = x[lcEventSrc_Scanner-1]
_ = x[lcEventSrc_Decom-2]
_ = x[lcEventSrc_Rebal-3]
_ = x[lcEventSrc_s3HeadObject-4]
_ = x[lcEventSrc_s3GetObject-5]
_ = x[lcEventSrc_s3ListObjects-6]
_ = x[lcEventSrc_s3PutObject-7]
_ = x[lcEventSrc_s3CopyObject-8]
_ = x[lcEventSrc_s3CompleteMultipartUpload-9]
}
const _lcEventSrc_name = "NoneScannerDecomRebals3HeadObjects3GetObjects3ListObjectss3PutObjects3CopyObjects3CompleteMultipartUpload"
var _lcEventSrc_index = [...]uint8{0, 4, 11, 16, 21, 33, 44, 57, 68, 80, 105}
func (i lcEventSrc) String() string {
if i >= lcEventSrc(len(_lcEventSrc_index)-1) {
return "lcEventSrc(" + strconv.FormatInt(int64(i), 10) + ")"
}
return _lcEventSrc_name[_lcEventSrc_index[i]:_lcEventSrc_index[i+1]]
}

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)
globalExpiryState.enqueueByDays(objInfo, evt, lcEventSrc_s3ListObjects)
if !evt.Action.DeleteRestored() {
continue
} // queue version for replication upon expired restored copies if needed.

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

@ -28,7 +28,6 @@ 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"
)
@ -61,7 +60,7 @@ type ObjectOptions struct {
DeleteReplication ReplicationState // Represents internal replication state needed for Delete replication
Transition TransitionOptions
Expiration ExpirationOptions
LifecycleEvent lifecycle.Event
LifecycleAuditEvent lcAuditEvent
WantChecksum *hash.Checksum // x-amz-checksum-XXX checksum sent to PutObject/ CompleteMultipartUpload.

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

@ -374,7 +374,8 @@ func deleteObjectVersions(ctx context.Context, o ObjectLayer, bucket string, toD
VersionID: dobj.VersionID,
}
traceFn := globalLifecycleSys.trace(oi)
tags := auditLifecycleTags(lcEvent)
// Note: NewerNoncurrentVersions action is performed only scanner today
tags := newLifecycleAuditEvent(lcEventSrc_Scanner, lcEvent).Tags()
// Send audit for the lifecycle delete operation
auditLogLifecycle(

8
cmd/object-handlers.go
View File

@ -479,7 +479,7 @@ func (api objectAPIHandlers) getObjectHandler(ctx context.Context, objectAPI Obj
event := evalActionFromLifecycle(ctx, *lc, rcfg, objInfo)
if event.Action.Delete() {
// apply whatever the expiry rule is.
applyExpiryRule(event, objInfo)
applyExpiryRule(event, lcEventSrc_s3GetObject, objInfo)
if !event.Action.DeleteRestored() {
// If the ILM action is not on restored object return error.
writeErrorResponseHeadersOnly(w, errorCodes.ToAPIErr(ErrNoSuchKey))
@ -733,7 +733,7 @@ func (api objectAPIHandlers) headObjectHandler(ctx context.Context, objectAPI Ob
event := evalActionFromLifecycle(ctx, *lc, rcfg, objInfo)
if event.Action.Delete() {
// apply whatever the expiry rule is.
applyExpiryRule(event, objInfo)
applyExpiryRule(event, lcEventSrc_s3HeadObject, objInfo)
if !event.Action.DeleteRestored() {
// If the ILM action is not on restored object return error.
writeErrorResponseHeadersOnly(w, errorCodes.ToAPIErr(ErrNoSuchKey))
@ -1560,7 +1560,7 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
if !remoteCallRequired && !globalTierConfigMgr.Empty() {
// Schedule object for immediate transition if eligible.
objInfo.ETag = origETag
enqueueTransitionImmediate(objInfo)
enqueueTransitionImmediate(objInfo, lcEventSrc_s3CopyObject)
// Remove the transitioned object whose object version is being overwritten.
logger.LogIf(ctx, os.Sweep())
}
@ -1937,7 +1937,7 @@ func (api objectAPIHandlers) PutObjectHandler(w http.ResponseWriter, r *http.Req
if !globalTierConfigMgr.Empty() {
// Schedule object for immediate transition if eligible.
objInfo.ETag = origETag
enqueueTransitionImmediate(objInfo)
enqueueTransitionImmediate(objInfo, lcEventSrc_s3PutObject)
logger.LogIf(ctx, os.Sweep())
}
}

2
cmd/object-multipart-handlers.go
View File

@ -1055,7 +1055,7 @@ func (api objectAPIHandlers) CompleteMultipartUploadHandler(w http.ResponseWrite
// Remove the transitioned object whose object version is being overwritten.
if !globalTierConfigMgr.Empty() {
// Schedule object for immediate transition if eligible.
enqueueTransitionImmediate(objInfo)
enqueueTransitionImmediate(objInfo, lcEventSrc_s3CompleteMultipartUpload)
logger.LogIf(ctx, os.Sweep())
}
}