frostfs-s3-gw/api/layer/versioning.go

305 lines
7.3 KiB
Go
Raw Normal View History

package layer
import (
"context"
"math"
"sort"
"strings"
"github.com/nspcc-dev/neofs-s3-gw/api/data"
)
type objectVersions struct {
name string
objects []*data.ObjectInfo
addList []string
delList []string
isSorted bool
}
func FromUnversioned() VersionOption {
return func(options *versionOptions) {
options.unversioned = true
}
}
type VersionOption func(*versionOptions)
type versionOptions struct {
unversioned bool
}
func formVersionOptions(opts ...VersionOption) *versionOptions {
options := &versionOptions{}
for _, opt := range opts {
opt(options)
}
return options
}
const (
VersionsDeleteMarkAttr = "S3-Versions-delete-mark"
DelMarkFullObject = "*"
unversionedObjectVersionID = "null"
objectSystemAttributeName = "S3-System-name"
attrVersionsIgnore = "S3-Versions-ignore"
versionsDelAttr = "S3-Versions-del"
versionsAddAttr = "S3-Versions-add"
versionsUnversionedAttr = "S3-Versions-unversioned"
)
func newObjectVersions(name string) *objectVersions {
return &objectVersions{name: name}
}
func (v *objectVersions) appendVersion(oi *data.ObjectInfo) {
delVers := splitVersions(oi.Headers[versionsDelAttr])
v.objects = append(v.objects, oi)
for _, del := range delVers {
if !contains(v.delList, del) {
v.delList = append(v.delList, del)
}
}
v.isSorted = false
}
func (v *objectVersions) sort() {
if !v.isSorted {
sort.Slice(v.objects, func(i, j int) bool {
o1, o2 := v.objects[i], v.objects[j]
if o1.CreationEpoch == o2.CreationEpoch {
l1, l2 := o1.Headers[versionsAddAttr], o2.Headers[versionsAddAttr]
if len(l1) != len(l2) {
if strings.HasPrefix(l1, l2) {
return false
} else if strings.HasPrefix(l2, l1) {
return true
}
}
return o1.Version() < o2.Version()
}
return o1.CreationEpoch < o2.CreationEpoch
})
v.formAddList()
v.isSorted = true
}
}
func (v *objectVersions) formAddList() {
for i := 0; i < len(v.objects); i++ {
var conflicts [][]string
for { // forming conflicts set (objects with the same creation epoch)
addVers := append(splitVersions(v.objects[i].Headers[versionsAddAttr]), v.objects[i].Version())
conflicts = append(conflicts, addVers)
if i == len(v.objects)-1 || v.objects[i].CreationEpoch != v.objects[i+1].CreationEpoch ||
containsVersions(v.objects[i+1], addVers) {
break
}
i++
}
if len(conflicts) == 1 {
v.addList = addIfNotContains(v.addList, conflicts[0])
continue
}
commonVersions, prevConflictedVersions, conflictedVersions := mergeVersionsConflicts(conflicts)
v.addList = commonVersions
v.addList = addIfNotContains(v.addList, prevConflictedVersions)
v.addList = addIfNotContains(v.addList, conflictedVersions)
}
}
func containsVersions(obj *data.ObjectInfo, versions []string) bool {
header := obj.Headers[versionsAddAttr]
for _, version := range versions {
if !strings.Contains(header, version) {
return false
}
}
return true
}
func addIfNotContains(list1, list2 []string) []string {
for _, add := range list2 {
if !contains(list1, add) {
list1 = append(list1, add)
}
}
return list1
}
func mergeVersionsConflicts(conflicts [][]string) ([]string, []string, []string) {
var currentVersions []string
var prevVersions []string
minLength := math.MaxInt32
for _, conflicted := range conflicts {
if len(conflicted)-1 < minLength {
minLength = len(conflicted) - 1
}
// last := conflicted[len(conflicted)-1]
// conflicts[j] = conflicted[:len(conflicted)-1]
// currentVersions = append(currentVersions, last)
}
var commonAddedVersions []string
diffIndex := 0
LOOP:
for k := 0; k < minLength; k++ {
candidate := conflicts[0][k]
for _, conflicted := range conflicts {
if conflicted[k] != candidate {
diffIndex = k
break LOOP
}
}
commonAddedVersions = append(commonAddedVersions, candidate)
}
for _, conflicted := range conflicts {
for j := diffIndex; j < len(conflicted); j++ {
prevVersions = append(prevVersions, conflicted[j])
}
}
sort.Strings(prevVersions)
sort.Strings(currentVersions)
return commonAddedVersions, prevVersions, currentVersions
}
func (v *objectVersions) isEmpty() bool {
return v == nil || len(v.objects) == 0
}
func (v *objectVersions) getLast(opts ...VersionOption) *data.ObjectInfo {
if v.isEmpty() {
return nil
}
options := formVersionOptions(opts...)
v.sort()
existedVersions := v.existedVersions()
for i := len(v.objects) - 1; i >= 0; i-- {
if contains(existedVersions, v.objects[i].Version()) {
delMarkHeader := v.objects[i].Headers[VersionsDeleteMarkAttr]
if delMarkHeader == "" {
if options.unversioned && v.objects[i].Headers[versionsUnversionedAttr] != "true" {
continue
}
return v.objects[i]
}
if delMarkHeader == DelMarkFullObject {
return nil
}
}
}
return nil
}
func (v *objectVersions) existedVersions() []string {
v.sort()
var res []string
for _, add := range v.addList {
if !contains(v.delList, add) {
res = append(res, add)
}
}
return res
}
func (v *objectVersions) getAddHeader() string {
v.sort()
return strings.Join(v.addList, ",")
}
func (n *layer) ListObjectVersions(ctx context.Context, p *ListObjectVersionsParams) (*ListObjectVersionsInfo, error) {
var (
allObjects = make([]*data.ObjectInfo, 0, p.MaxKeys)
res = &ListObjectVersionsInfo{}
)
versions, err := n.getAllObjectsVersions(ctx, p.BktInfo, p.Prefix, p.Delimiter)
if err != nil {
return nil, err
}
sortedNames := make([]string, 0, len(versions))
for k := range versions {
sortedNames = append(sortedNames, k)
}
sort.Strings(sortedNames)
for _, name := range sortedNames {
sortedVersions := versions[name]
sort.Slice(sortedVersions, func(i, j int) bool {
return sortedVersions[j].NodeVersion.Timestamp < sortedVersions[i].NodeVersion.Timestamp // sort in reverse order
})
for _, version := range sortedVersions {
allObjects = append(allObjects, version.ObjectInfo)
}
}
for i, obj := range allObjects {
if obj.Name >= p.KeyMarker && obj.Version() >= p.VersionIDMarker {
allObjects = allObjects[i:]
break
}
}
res.CommonPrefixes, allObjects = triageObjects(allObjects)
if len(allObjects) > p.MaxKeys {
res.IsTruncated = true
res.NextKeyMarker = allObjects[p.MaxKeys].Name
res.NextVersionIDMarker = allObjects[p.MaxKeys].Version()
allObjects = allObjects[:p.MaxKeys]
res.KeyMarker = allObjects[p.MaxKeys-1].Name
res.VersionIDMarker = allObjects[p.MaxKeys-1].Version()
}
objects := make([]*ObjectVersionInfo, len(allObjects))
for i, obj := range allObjects {
objects[i] = &ObjectVersionInfo{Object: obj}
if i == 0 || allObjects[i-1].Name != obj.Name {
objects[i].IsLatest = true
}
}
res.Version, res.DeleteMarker = triageVersions(objects)
return res, nil
}
func triageVersions(objVersions []*ObjectVersionInfo) ([]*ObjectVersionInfo, []*ObjectVersionInfo) {
if len(objVersions) == 0 {
return nil, nil
}
var resVersion []*ObjectVersionInfo
var resDelMarkVersions []*ObjectVersionInfo
for _, version := range objVersions {
if version.Object.IsDeleteMarker {
resDelMarkVersions = append(resDelMarkVersions, version)
} else {
resVersion = append(resVersion, version)
}
}
return resVersion, resDelMarkVersions
}
func contains(list []string, elem string) bool {
for _, item := range list {
if elem == item {
return true
}
}
return false
}