frostfs-node/pkg/local_object_storage/metabase/graveyard_test.go

438 lines
11 KiB
Go
Raw Permalink Normal View History

package meta_test
import (
"context"
"testing"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/object"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/internal/testutil"
meta "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/metabase"
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
oidtest "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id/test"
"github.com/stretchr/testify/require"
)
func TestDB_IterateDeletedObjects_EmptyDB(t *testing.T) {
db := newDB(t)
var counter int
var iterGravePRM meta.GraveyardIterationPrm
iterGravePRM.SetHandler(func(garbage meta.TombstonedObject) error {
counter++
return nil
})
err := db.IterateOverGraveyard(iterGravePRM)
require.NoError(t, err)
require.Zero(t, counter)
var iterGCPRM meta.GarbageIterationPrm
iterGCPRM.SetHandler(func(garbage meta.GarbageObject) error {
counter++
return nil
})
err = db.IterateOverGarbage(iterGCPRM)
require.NoError(t, err)
require.Zero(t, counter)
}
func TestDB_Iterate_OffsetNotFound(t *testing.T) {
db := newDB(t)
obj1 := testutil.GenerateObject()
obj2 := testutil.GenerateObject()
var addr1 oid.Address
err := addr1.DecodeString("AUSF6rhReoAdPVKYUZWW9o2LbtTvekn54B3JXi7pdzmn/2daLhLB7yVXbjBaKkckkuvjX22BxRYuSHy9RPxuH9PZS")
require.NoError(t, err)
var addr2 oid.Address
err = addr2.DecodeString("CwYYr6sFLU1zK6DeBTVd8SReADUoxYobUhSrxgXYxCVn/ANYbnJoQqdjmU5Dhk3LkxYj5E9nJHQFf8LjTEcap9TxM")
require.NoError(t, err)
var addr3 oid.Address
err = addr3.DecodeString("6ay4GfhR9RgN28d5ufg63toPetkYHGcpcW7G3b7QWSek/ANYbnJoQqdjmU5Dhk3LkxYj5E9nJHQFf8LjTEcap9TxM")
require.NoError(t, err)
obj1.SetContainerID(addr1.Container())
obj1.SetID(addr1.Object())
obj2.SetContainerID(addr2.Container())
obj2.SetID(addr2.Object())
err = putBig(db, obj1)
require.NoError(t, err)
var inhumePrm meta.InhumePrm
inhumePrm.SetAddresses(object.AddressOf(obj1))
inhumePrm.SetGCMark()
_, err = db.Inhume(context.Background(), inhumePrm)
require.NoError(t, err)
var counter int
var iterGCPRM meta.GarbageIterationPrm
iterGCPRM.SetOffset(object.AddressOf(obj2))
iterGCPRM.SetHandler(func(garbage meta.GarbageObject) error {
require.Equal(t, garbage.Address(), addr1)
counter++
return nil
})
err = db.IterateOverGarbage(iterGCPRM)
require.NoError(t, err)
// the second object would be put after the
// first, so it is expected that iteration
// will not receive the first object
require.Equal(t, 0, counter)
iterGCPRM.SetOffset(addr3)
iterGCPRM.SetHandler(func(garbage meta.GarbageObject) error {
require.Equal(t, garbage.Address(), addr1)
counter++
return nil
})
err = db.IterateOverGarbage(iterGCPRM)
require.NoError(t, err)
// the third object would be put before the
// first, so it is expected that iteration
// will receive the first object
require.Equal(t, 1, counter)
}
func TestDB_IterateDeletedObjects(t *testing.T) {
db := newDB(t)
// generate and put 4 objects
obj1 := testutil.GenerateObject()
obj2 := testutil.GenerateObject()
obj3 := testutil.GenerateObject()
obj4 := testutil.GenerateObject()
var err error
err = putBig(db, obj1)
require.NoError(t, err)
err = putBig(db, obj2)
require.NoError(t, err)
err = putBig(db, obj3)
require.NoError(t, err)
err = putBig(db, obj4)
require.NoError(t, err)
var inhumePrm meta.InhumePrm
// inhume with tombstone
addrTombstone := oidtest.Address()
inhumePrm.SetAddresses(object.AddressOf(obj1), object.AddressOf(obj2))
inhumePrm.SetTombstoneAddress(addrTombstone)
_, err = db.Inhume(context.Background(), inhumePrm)
require.NoError(t, err)
inhumePrm.SetAddresses(object.AddressOf(obj3), object.AddressOf(obj4))
inhumePrm.SetGCMark()
// inhume with GC mark
_, err = db.Inhume(context.Background(), inhumePrm)
require.NoError(t, err)
var (
counterAll int
buriedTS, buriedGC []oid.Address
)
var iterGravePRM meta.GraveyardIterationPrm
iterGravePRM.SetHandler(func(tomstoned meta.TombstonedObject) error {
require.Equal(t, addrTombstone, tomstoned.Tombstone())
buriedTS = append(buriedTS, tomstoned.Address())
counterAll++
return nil
})
err = db.IterateOverGraveyard(iterGravePRM)
require.NoError(t, err)
var iterGCPRM meta.GarbageIterationPrm
iterGCPRM.SetHandler(func(garbage meta.GarbageObject) error {
buriedGC = append(buriedGC, garbage.Address())
counterAll++
return nil
})
err = db.IterateOverGarbage(iterGCPRM)
require.NoError(t, err)
// objects covered with a tombstone
// also receive GS mark
garbageExpected := []oid.Address{
object.AddressOf(obj1), object.AddressOf(obj2),
object.AddressOf(obj3), object.AddressOf(obj4),
}
graveyardExpected := []oid.Address{
object.AddressOf(obj1), object.AddressOf(obj2),
}
require.Equal(t, len(garbageExpected)+len(graveyardExpected), counterAll)
require.ElementsMatch(t, graveyardExpected, buriedTS)
require.ElementsMatch(t, garbageExpected, buriedGC)
}
func TestDB_IterateOverGraveyard_Offset(t *testing.T) {
db := newDB(t)
// generate and put 4 objects
obj1 := testutil.GenerateObject()
obj2 := testutil.GenerateObject()
obj3 := testutil.GenerateObject()
obj4 := testutil.GenerateObject()
var err error
err = putBig(db, obj1)
require.NoError(t, err)
err = putBig(db, obj2)
require.NoError(t, err)
err = putBig(db, obj3)
require.NoError(t, err)
err = putBig(db, obj4)
require.NoError(t, err)
// inhume with tombstone
addrTombstone := oidtest.Address()
var inhumePrm meta.InhumePrm
inhumePrm.SetAddresses(
object.AddressOf(obj1), object.AddressOf(obj2),
object.AddressOf(obj3), object.AddressOf(obj4))
inhumePrm.SetTombstoneAddress(addrTombstone)
_, err = db.Inhume(context.Background(), inhumePrm)
require.NoError(t, err)
expectedGraveyard := []oid.Address{
object.AddressOf(obj1), object.AddressOf(obj2),
object.AddressOf(obj3), object.AddressOf(obj4),
}
var (
counter int
firstIterationSize = len(expectedGraveyard) / 2
gotGraveyard []oid.Address
)
var iterGraveyardPrm meta.GraveyardIterationPrm
iterGraveyardPrm.SetHandler(func(tombstoned meta.TombstonedObject) error {
require.Equal(t, addrTombstone, tombstoned.Tombstone())
gotGraveyard = append(gotGraveyard, tombstoned.Address())
counter++
if counter == firstIterationSize {
return meta.ErrInterruptIterator
}
return nil
})
err = db.IterateOverGraveyard(iterGraveyardPrm)
require.NoError(t, err)
require.Equal(t, firstIterationSize, counter)
require.Equal(t, firstIterationSize, len(gotGraveyard))
// last received address is an offset
offset := gotGraveyard[len(gotGraveyard)-1]
iterGraveyardPrm.SetOffset(offset)
iterGraveyardPrm.SetHandler(func(tombstoned meta.TombstonedObject) error {
require.Equal(t, addrTombstone, tombstoned.Tombstone())
gotGraveyard = append(gotGraveyard, tombstoned.Address())
counter++
return nil
})
err = db.IterateOverGraveyard(iterGraveyardPrm)
require.NoError(t, err)
require.Equal(t, len(expectedGraveyard), counter)
require.ElementsMatch(t, gotGraveyard, expectedGraveyard)
// last received object (last in db) as offset
// should lead to no iteration at all
offset = gotGraveyard[len(gotGraveyard)-1]
iterGraveyardPrm.SetOffset(offset)
iWasCalled := false
iterGraveyardPrm.SetHandler(func(tombstoned meta.TombstonedObject) error {
iWasCalled = true
return nil
})
err = db.IterateOverGraveyard(iterGraveyardPrm)
require.NoError(t, err)
require.False(t, iWasCalled)
}
func TestDB_IterateOverGarbage_Offset(t *testing.T) {
db := newDB(t)
// generate and put 4 objects
obj1 := testutil.GenerateObject()
obj2 := testutil.GenerateObject()
obj3 := testutil.GenerateObject()
obj4 := testutil.GenerateObject()
var err error
err = putBig(db, obj1)
require.NoError(t, err)
err = putBig(db, obj2)
require.NoError(t, err)
err = putBig(db, obj3)
require.NoError(t, err)
err = putBig(db, obj4)
require.NoError(t, err)
var inhumePrm meta.InhumePrm
inhumePrm.SetAddresses(
object.AddressOf(obj1), object.AddressOf(obj2),
object.AddressOf(obj3), object.AddressOf(obj4))
inhumePrm.SetGCMark()
_, err = db.Inhume(context.Background(), inhumePrm)
require.NoError(t, err)
expectedGarbage := []oid.Address{
object.AddressOf(obj1), object.AddressOf(obj2),
object.AddressOf(obj3), object.AddressOf(obj4),
}
var (
counter int
firstIterationSize = len(expectedGarbage) / 2
gotGarbage []oid.Address
)
var iterGarbagePrm meta.GarbageIterationPrm
iterGarbagePrm.SetHandler(func(garbage meta.GarbageObject) error {
gotGarbage = append(gotGarbage, garbage.Address())
counter++
if counter == firstIterationSize {
return meta.ErrInterruptIterator
}
return nil
})
err = db.IterateOverGarbage(iterGarbagePrm)
require.NoError(t, err)
require.Equal(t, firstIterationSize, counter)
require.Equal(t, firstIterationSize, len(gotGarbage))
// last received address is an offset
offset := gotGarbage[len(gotGarbage)-1]
iterGarbagePrm.SetOffset(offset)
iterGarbagePrm.SetHandler(func(garbage meta.GarbageObject) error {
gotGarbage = append(gotGarbage, garbage.Address())
counter++
return nil
})
err = db.IterateOverGarbage(iterGarbagePrm)
require.NoError(t, err)
require.Equal(t, len(expectedGarbage), counter)
require.ElementsMatch(t, gotGarbage, expectedGarbage)
// last received object (last in db) as offset
// should lead to no iteration at all
offset = gotGarbage[len(gotGarbage)-1]
iterGarbagePrm.SetOffset(offset)
iWasCalled := false
iterGarbagePrm.SetHandler(func(garbage meta.GarbageObject) error {
iWasCalled = true
return nil
})
err = db.IterateOverGarbage(iterGarbagePrm)
require.NoError(t, err)
require.False(t, iWasCalled)
}
func TestDB_DropGraves(t *testing.T) {
db := newDB(t)
// generate and put 2 objects
obj1 := testutil.GenerateObject()
obj2 := testutil.GenerateObject()
var err error
err = putBig(db, obj1)
require.NoError(t, err)
err = putBig(db, obj2)
require.NoError(t, err)
// inhume with tombstone
addrTombstone := oidtest.Address()
var inhumePrm meta.InhumePrm
inhumePrm.SetAddresses(object.AddressOf(obj1), object.AddressOf(obj2))
inhumePrm.SetTombstoneAddress(addrTombstone)
_, err = db.Inhume(context.Background(), inhumePrm)
require.NoError(t, err)
buriedTS := make([]meta.TombstonedObject, 0)
var iterGravePRM meta.GraveyardIterationPrm
var counter int
iterGravePRM.SetHandler(func(tomstoned meta.TombstonedObject) error {
buriedTS = append(buriedTS, tomstoned)
counter++
return nil
})
err = db.IterateOverGraveyard(iterGravePRM)
require.NoError(t, err)
require.Equal(t, 2, counter)
err = db.DropGraves(buriedTS)
require.NoError(t, err)
counter = 0
iterGravePRM.SetHandler(func(_ meta.TombstonedObject) error {
counter++
return nil
})
err = db.IterateOverGraveyard(iterGravePRM)
require.NoError(t, err)
require.Zero(t, counter)
}