mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-19 20:44:28 +02:00
* s3 lifecycle: share the daily bucket walk across shards The daily pass invoked the walker once per shard, and every shard's walk listed the full bucket tree while keeping only its own 1/16 of the entries — 16 full listings per pass by default (#11791). Run now groups the shards by the walk branch they'll take (recovery vs steady, using the same cursor data and hash checks runShard applies) and starts one shared walk per group, so a pass lists each bucket at most twice. WalkerFunc and WalkBuckets take a shard set instead of a single shard. runShard keeps its own decision and waits on its group's walk; a shard whose freshly-loaded cursor disagrees with the plan falls back to a per-shard walk, so mid-pass cursor rewrites can't under-cover a shard. * s3 lifecycle: skip the .versions/ pre-pass on non-versioned buckets walkBucketTree listed every directory twice — once looking for .versions/ dirs, once for regular entries. On a bucket that never had versioning enabled the first pass can only find coincidentally-named user folders, so the extra listing is pure cost (#11791). FilerListFunc now takes the walk view and consults Snapshot.BucketVersioned; non-versioned buckets skip pass 1 entirely. A *-suffixed .versions user folder on such a bucket is recursed into as a regular directory so real objects inside still surface; a nil view keeps the two-pass behavior. * s3 lifecycle: prune walk directories to active rule prefixes The daily walk listed every directory under a bucket even when no rule prefix could reach it. Collect the active rule prefixes of each bucket from the walk view (new Snapshot.BucketPrefixes) and skip recursing into directories whose keys can never match; a bucket with no active actions is skipped entirely, saving the listing. .uploads stays in scope regardless of prefixes because AbortMPU rules match DestKey, and .versions expansion is pruned only conservatively (the logical key is a string-prefix of the dir key). * s3 lifecycle: join planned walks before Run returns planWalks starts a shared walk goroutine per group, but a shard whose authoritative cursor load then fails exits runShard without waiting on it. Run returned while the walk still listed or dispatched deletes against connections the caller closes on return. Join every planned walk after the shard waitgroup. * s3 lifecycle: treat object-locked buckets as versioned in the walk Object Lock requires versioning and every writer sets both extended attributes, but IsBucketVersioned only read ExtVersioningKey. A locked bucket missing that attribute would have its .versions dirs walked as ordinary folders, hiding version siblings from lifecycle. Classify ExtObjectLockEnabledKey as versioned too.
157 lines
4.2 KiB
Go
157 lines
4.2 KiB
Go
// Package engine compiles per-bucket lifecycle rules into a CompiledAction
|
|
// snapshot. One XML <Rule> expands into N compiled actions (one per populated
|
|
// action sub-element); each is keyed by ActionKey{bucket, rule_hash,
|
|
// action_kind} so sibling actions schedule and degrade independently.
|
|
package engine
|
|
|
|
import (
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
|
|
)
|
|
|
|
type Engine struct {
|
|
current atomic.Pointer[Snapshot]
|
|
}
|
|
|
|
func New() *Engine {
|
|
e := &Engine{}
|
|
e.current.Store(&Snapshot{actions: map[s3lifecycle.ActionKey]*CompiledAction{}})
|
|
return e
|
|
}
|
|
|
|
func (e *Engine) Snapshot() *Snapshot { return e.current.Load() }
|
|
|
|
// Snapshot fields are append-only after Compile except CompiledAction.engineState,
|
|
// which transitions inactive -> active atomically via markActive.
|
|
type Snapshot struct {
|
|
id uint64
|
|
buckets map[string]*BucketIndex
|
|
actions map[s3lifecycle.ActionKey]*CompiledAction
|
|
|
|
allActionsSorted []*CompiledAction
|
|
|
|
// Routing indexes hold every ActionKey by mode regardless of activation;
|
|
// dispatch filters on IsActive().
|
|
originalDelayGroups map[time.Duration][]s3lifecycle.ActionKey
|
|
predicateActions []s3lifecycle.ActionKey
|
|
dateActions map[s3lifecycle.ActionKey]time.Time
|
|
}
|
|
|
|
func (s *Snapshot) SnapshotID() uint64 { return s.id }
|
|
|
|
type BucketIndex struct {
|
|
bucket string
|
|
versioned bool
|
|
actionKeys []s3lifecycle.ActionKey
|
|
}
|
|
|
|
type CompiledAction struct {
|
|
Rule *s3lifecycle.Rule
|
|
Bucket string
|
|
Key s3lifecycle.ActionKey
|
|
Delay time.Duration
|
|
PredicateSensitive bool
|
|
Mode RuleMode
|
|
engineState atomic.Uint32
|
|
}
|
|
|
|
const (
|
|
engineStateInactive uint32 = 0
|
|
engineStateActive uint32 = 1
|
|
)
|
|
|
|
func (a *CompiledAction) IsActive() bool {
|
|
return a.engineState.Load() == engineStateActive
|
|
}
|
|
|
|
func (a *CompiledAction) markActive() {
|
|
a.engineState.Store(engineStateActive)
|
|
}
|
|
|
|
// MarkActive mirrors the in-memory hint after a durable bootstrap_complete
|
|
// write. No-op if the key isn't in the snapshot.
|
|
func (s *Snapshot) MarkActive(key s3lifecycle.ActionKey) {
|
|
if a, ok := s.actions[key]; ok {
|
|
a.markActive()
|
|
}
|
|
}
|
|
|
|
func (s *Snapshot) Action(key s3lifecycle.ActionKey) *CompiledAction {
|
|
return s.actions[key]
|
|
}
|
|
|
|
// AllActions: caller must not mutate.
|
|
func (s *Snapshot) AllActions() []*CompiledAction {
|
|
return s.allActionsSorted
|
|
}
|
|
|
|
// OriginalDelayGroups / PredicateActions / DateActions return defensive copies.
|
|
func (s *Snapshot) OriginalDelayGroups() map[time.Duration][]s3lifecycle.ActionKey {
|
|
out := make(map[time.Duration][]s3lifecycle.ActionKey, len(s.originalDelayGroups))
|
|
for d, keys := range s.originalDelayGroups {
|
|
copied := make([]s3lifecycle.ActionKey, len(keys))
|
|
copy(copied, keys)
|
|
out[d] = copied
|
|
}
|
|
return out
|
|
}
|
|
|
|
func (s *Snapshot) PredicateActions() []s3lifecycle.ActionKey {
|
|
out := make([]s3lifecycle.ActionKey, len(s.predicateActions))
|
|
copy(out, s.predicateActions)
|
|
return out
|
|
}
|
|
|
|
func (s *Snapshot) DateActions() map[s3lifecycle.ActionKey]time.Time {
|
|
out := make(map[s3lifecycle.ActionKey]time.Time, len(s.dateActions))
|
|
for k, v := range s.dateActions {
|
|
out[k] = v
|
|
}
|
|
return out
|
|
}
|
|
|
|
func (s *Snapshot) BucketVersioned(bucket string) bool {
|
|
if bi, ok := s.buckets[bucket]; ok {
|
|
return bi.versioned
|
|
}
|
|
return false
|
|
}
|
|
|
|
// BucketPrefixes returns the distinct rule prefixes of the bucket's
|
|
// active actions in this view. An empty result means no active action
|
|
// can match any entry in the bucket; a "" element means some rule
|
|
// matches every key.
|
|
func (s *Snapshot) BucketPrefixes(bucket string) []string {
|
|
bi := s.buckets[bucket]
|
|
if bi == nil {
|
|
return nil
|
|
}
|
|
seen := map[string]bool{}
|
|
var out []string
|
|
for _, k := range bi.actionKeys {
|
|
a := s.actions[k]
|
|
if a == nil || !a.IsActive() || a.Mode == ModeDisabled || a.Rule == nil {
|
|
continue
|
|
}
|
|
if !seen[a.Rule.Prefix] {
|
|
seen[a.Rule.Prefix] = true
|
|
out = append(out, a.Rule.Prefix)
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// BucketActionKeys returns a defensive copy of the action keys for bucket,
|
|
// or nil if the bucket has no compiled actions in this snapshot.
|
|
func (s *Snapshot) BucketActionKeys(bucket string) []s3lifecycle.ActionKey {
|
|
bi, ok := s.buckets[bucket]
|
|
if !ok {
|
|
return nil
|
|
}
|
|
out := make([]s3lifecycle.ActionKey, len(bi.actionKeys))
|
|
copy(out, bi.actionKeys)
|
|
return out
|
|
}
|