Skip to content

[ILM] Relocate hot objects across server pools by GET frequency #60

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mrjavadseydi wants to merge 1 commit into
base: main
Choose a base branch
from
Open

[ILM] Relocate hot objects across server pools by GET frequency #60

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions cmd/bucket-lifecycle-audit.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ const (
lcEventSrc_s3PutObject
lcEventSrc_s3CopyObject
lcEventSrc_s3CompleteMultipartUpload
lcEventSrc_AccessTier
)

//revive:enable:var-naming
Expand Down
5 changes: 4 additions & 1 deletion cmd/config-current.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -227,7 +227,7 @@ func initHelp() {
},
config.HelpKV{
Key: config.ILMSubSys,
Description: "manage ILM settings for expiration and transition workers",
Description: "manage ILM settings for expiration, transition, and access-tier workers",
Optional: true,
},
}
Expand Down Expand Up @@ -704,6 +704,9 @@ func applyDynamicConfigForSubSys(ctx context.Context, objAPI ObjectLayer, s conf
if globalExpiryState != nil {
globalExpiryState.ResizeWorkers(ilmCfg.ExpirationWorkers)
}
if globalAccessTierState != nil {
globalAccessTierState.UpdateWorkers(ilmCfg.AccessWorkers)
}
globalILMConfig.update(ilmCfg)
}
}
Expand Down
14 changes: 14 additions & 0 deletions cmd/data-scanner.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -899,6 +899,7 @@ type scannerItem struct {
objectName string // Only the object name without prefixes.
replication replicationConfig
lifeCycle *lifecycle.Lifecycle
poolIdx int
Typ fs.FileMode
heal struct {
enabled bool
Expand All @@ -909,6 +910,7 @@ type scannerItem struct {

type sizeSummary struct {
totalSize int64
hotTierSize int64
versions uint64
deleteMarkers uint64
replicatedSize int64
Expand Down Expand Up @@ -1159,6 +1161,14 @@ eventLoop:
globalExpiryState.enqueueNoncurrentVersions(i.bucket, toDel, noncurrentEvents)
}
i.alertExcessiveVersions(remainingVersions, cumulativeSize)
if globalILMConfig.accessTieringEnabled() {
for idx, oi := range objInfos {
if oi.IsLatest && events[idx].Action == lifecycle.NoneAction {
applyAccessTransition(ctx, i, oi)
break
}
}
}
}

func evalActionFromLifecycle(ctx context.Context, lc lifecycle.Lifecycle, lr lock.Retention, rcfg *replication.Config, obj ObjectInfo) lifecycle.Event {
Expand Down Expand Up @@ -1474,6 +1484,8 @@ const (
ILMFreeVersionDelete = "ilm:free-version-delete"
// ILMTransition - audit trail for ILM transitioning.
ILMTransition = " ilm:transition"
// ILMAccessTier - audit trail for moving objects between server pools.
ILMAccessTier = "ilm:access-tier"
)

func auditLogLifecycle(ctx context.Context, oi ObjectInfo, event string, tags map[string]string, traceFn func(event string, metadata map[string]string, err error)) {
Expand All @@ -1485,6 +1497,8 @@ func auditLogLifecycle(ctx context.Context, oi ObjectInfo, event string, tags ma
apiName = "ILMFreeVersionDelete"
case ILMTransition:
apiName = "ILMTransition"
case ILMAccessTier:
apiName = "ILMAccessTier"
}
auditLogInternal(ctx, AuditLogOptions{
Event: event,
Expand Down
65 changes: 60 additions & 5 deletions cmd/data-usage-cache.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@ type dataUsageEntry struct {
Children dataUsageHashMap `msg:"ch"`
// These fields do no include any children.
Size int64 `msg:"sz"`
HotTierSize int64 `msg:"hts"`
Objects uint64 `msg:"os"`
Versions uint64 `msg:"vs"` // Versions that are not delete markers.
DeleteMarkers uint64 `msg:"dms"`
Expand Down Expand Up @@ -134,8 +135,8 @@ func (ts tierStats) add(u tierStats) tierStats {
}
}

//msgp:encode ignore dataUsageEntryV2 dataUsageEntryV3 dataUsageEntryV4 dataUsageEntryV5 dataUsageEntryV6 dataUsageEntryV7
//msgp:marshal ignore dataUsageEntryV2 dataUsageEntryV3 dataUsageEntryV4 dataUsageEntryV5 dataUsageEntryV6 dataUsageEntryV7
//msgp:encode ignore dataUsageEntryV2 dataUsageEntryV3 dataUsageEntryV4 dataUsageEntryV5 dataUsageEntryV6 dataUsageEntryV7 dataUsageEntryV8
//msgp:marshal ignore dataUsageEntryV2 dataUsageEntryV3 dataUsageEntryV4 dataUsageEntryV5 dataUsageEntryV6 dataUsageEntryV7 dataUsageEntryV8

//msgp:tuple dataUsageEntryV2
type dataUsageEntryV2 struct {
Expand Down Expand Up @@ -199,14 +200,30 @@ type dataUsageEntryV7 struct {
Compacted bool `msg:"c"`
}

// dataUsageEntryV8 is the on-disk shape before access-tier accounting was
// introduced. Keep it so caches written by the previous release decode
// without being discarded.
type dataUsageEntryV8 struct {
Children dataUsageHashMap `msg:"ch"`
// These fields do no include any children.
Size int64 `msg:"sz"`
Objects uint64 `msg:"os"`
Versions uint64 `msg:"vs"`
DeleteMarkers uint64 `msg:"dms"`
ObjSizes sizeHistogram `msg:"szs"`
ObjVersions versionsHistogram `msg:"vh"`
AllTierStats *allTierStats `msg:"ats,omitempty"`
Compacted bool `msg:"c"`
}

// dataUsageCache contains a cache of data usage entries latest version.
type dataUsageCache struct {
Info dataUsageCacheInfo
Cache map[string]dataUsageEntry
}

//msgp:encode ignore dataUsageCacheV2 dataUsageCacheV3 dataUsageCacheV4 dataUsageCacheV5 dataUsageCacheV6 dataUsageCacheV7
//msgp:marshal ignore dataUsageCacheV2 dataUsageCacheV3 dataUsageCacheV4 dataUsageCacheV5 dataUsageCacheV6 dataUsageCacheV7
//msgp:encode ignore dataUsageCacheV2 dataUsageCacheV3 dataUsageCacheV4 dataUsageCacheV5 dataUsageCacheV6 dataUsageCacheV7 dataUsageCacheV8
//msgp:marshal ignore dataUsageCacheV2 dataUsageCacheV3 dataUsageCacheV4 dataUsageCacheV5 dataUsageCacheV6 dataUsageCacheV7 dataUsageCacheV8

// dataUsageCacheV2 contains a cache of data usage entries version 2.
type dataUsageCacheV2 struct {
Expand Down Expand Up @@ -244,6 +261,12 @@ type dataUsageCacheV7 struct {
Cache map[string]dataUsageEntryV7
}

// dataUsageCacheV8 contains a cache of data usage entries version 8.
type dataUsageCacheV8 struct {
Info dataUsageCacheInfo
Cache map[string]dataUsageEntryV8
}

//msgp:ignore dataUsageEntryInfo
type dataUsageEntryInfo struct {
Name string
Expand Down Expand Up @@ -272,6 +295,7 @@ type dataUsageCacheInfo struct {

func (e *dataUsageEntry) addSizes(summary sizeSummary) {
e.Size += summary.totalSize
e.HotTierSize += summary.hotTierSize
e.Versions += summary.versions
e.DeleteMarkers += summary.deleteMarkers
e.ObjSizes.add(summary.totalSize)
Expand All @@ -291,6 +315,7 @@ func (e *dataUsageEntry) merge(other dataUsageEntry) {
e.Versions += other.Versions
e.DeleteMarkers += other.DeleteMarkers
e.Size += other.Size
e.HotTierSize += other.HotTierSize

for i, v := range other.ObjSizes[:] {
e.ObjSizes[i] += v
Expand Down Expand Up @@ -431,6 +456,7 @@ func (d *dataUsageCache) dui(path string, buckets []BucketInfo) DataUsageInfo {
flat := d.flatten(*e)
dui := DataUsageInfo{
LastUpdate: d.Info.LastUpdate,
ScannerCycle: d.Info.NextCycle,
ObjectsTotalCount: flat.Objects,
VersionsTotalCount: flat.Versions,
DeleteMarkersTotalCount: flat.DeleteMarkers,
Expand Down Expand Up @@ -781,6 +807,7 @@ func (d *dataUsageCache) bucketsUsageInfo(buckets []BucketInfo) map[string]Bucke
flat := d.flatten(*e)
bui := BucketUsageInfo{
Size: uint64(flat.Size),
HotTierSize: uint64(max(flat.HotTierSize, 0)),
VersionsCount: flat.Versions,
ObjectsCount: flat.Objects,
DeleteMarkersCount: flat.DeleteMarkers,
Expand Down Expand Up @@ -980,7 +1007,8 @@ func (d *dataUsageCache) save(ctx context.Context, store objectIO, name string)
// Bumping the cache version will drop data from previous versions
// and write new data with the new version.
const (
dataUsageCacheVerCurrent = 8
dataUsageCacheVerCurrent = 9
dataUsageCacheVerV8 = 8
dataUsageCacheVerV7 = 7
dataUsageCacheVerV6 = 6
dataUsageCacheVerV5 = 5
Expand Down Expand Up @@ -1181,6 +1209,33 @@ func (d *dataUsageCache) deserialize(r io.Reader) error {
}
}

return nil
case dataUsageCacheVerV8:
// Zstd compressed.
dec, err := zstd.NewReader(r, zstd.WithDecoderConcurrency(2))
if err != nil {
return err
}
defer dec.Close()
dold := &dataUsageCacheV8{}
if err = dold.DecodeMsg(msgp.NewReader(dec)); err != nil {
return err
}
d.Info = dold.Info
d.Cache = make(map[string]dataUsageEntry, len(dold.Cache))
for k, v := range dold.Cache {
d.Cache[k] = dataUsageEntry{
Children: v.Children,
Size: v.Size,
Objects: v.Objects,
Versions: v.Versions,
DeleteMarkers: v.DeleteMarkers,
ObjSizes: v.ObjSizes,
ObjVersions: v.ObjVersions,
AllTierStats: v.AllTierStats,
Compacted: v.Compacted,
}
}
return nil
case dataUsageCacheVerCurrent:
// Zstd compressed.
Expand Down
Loading