package writecache import ( "os" "path/filepath" "testing" "time" objectCore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/object" "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor" "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/common" "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/fstree" meta "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/metabase" "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard/mode" "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger" checksumtest "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/checksum/test" apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status" cidtest "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id/test" "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object" oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id" oidtest "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id/test" usertest "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/user/test" versionSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/version" "github.com/stretchr/testify/require" "go.etcd.io/bbolt" "go.uber.org/atomic" "go.uber.org/zap/zaptest" ) type objectPair struct { addr oid.Address obj *object.Object } func TestFlush(t *testing.T) { const ( objCount = 4 smallSize = 256 ) newCache := func(t *testing.T, opts ...Option) (Cache, *blobstor.BlobStor, *meta.DB) { dir := t.TempDir() mb := meta.New( meta.WithPath(filepath.Join(dir, "meta")), meta.WithEpochState(dummyEpoch{})) require.NoError(t, mb.Open(false)) require.NoError(t, mb.Init()) bs := blobstor.New(blobstor.WithStorages([]blobstor.SubStorage{ { Storage: fstree.New( fstree.WithPath(filepath.Join(dir, "blob")), fstree.WithDepth(0), fstree.WithDirNameLen(1)), }, })) require.NoError(t, bs.Open(false)) require.NoError(t, bs.Init()) wc := New( append([]Option{ WithLogger(&logger.Logger{Logger: zaptest.NewLogger(t)}), WithPath(filepath.Join(dir, "writecache")), WithSmallObjectSize(smallSize), WithMetabase(mb), WithBlobstor(bs), }, opts...)...) require.NoError(t, wc.Open(false)) initWC(t, wc) // First set mode for metabase and blobstor to prevent background flushes. require.NoError(t, mb.SetMode(mode.ReadOnly)) require.NoError(t, bs.SetMode(mode.ReadOnly)) return wc, bs, mb } putObjects := func(t *testing.T, c Cache) []objectPair { objects := make([]objectPair, objCount) for i := range objects { objects[i] = putObject(t, c, 1+(i%2)*smallSize) } return objects } check := func(t *testing.T, mb *meta.DB, bs *blobstor.BlobStor, objects []objectPair) { for i := range objects { var mPrm meta.StorageIDPrm mPrm.SetAddress(objects[i].addr) mRes, err := mb.StorageID(mPrm) require.NoError(t, err) var prm common.GetPrm prm.Address = objects[i].addr prm.StorageID = mRes.StorageID() res, err := bs.Get(prm) require.NoError(t, err) require.Equal(t, objects[i].obj, res.Object) } } t.Run("no errors", func(t *testing.T) { wc, bs, mb := newCache(t) objects := putObjects(t, wc) require.NoError(t, bs.SetMode(mode.ReadWrite)) require.NoError(t, mb.SetMode(mode.ReadWrite)) wc.(*cache).flushed.Add(objects[0].addr.EncodeToString(), true) wc.(*cache).flushed.Add(objects[1].addr.EncodeToString(), false) require.NoError(t, wc.Flush(false)) for i := 0; i < 2; i++ { var mPrm meta.GetPrm mPrm.SetAddress(objects[i].addr) _, err := mb.Get(mPrm) require.Error(t, err) _, err = bs.Get(common.GetPrm{Address: objects[i].addr}) require.Error(t, err) } check(t, mb, bs, objects[2:]) }) t.Run("flush on moving to degraded mode", func(t *testing.T) { wc, bs, mb := newCache(t) objects := putObjects(t, wc) // Blobstor is read-only, so we expect en error from `flush` here. require.Error(t, wc.SetMode(mode.Degraded)) // First move to read-only mode to close background workers. require.NoError(t, wc.SetMode(mode.ReadOnly)) require.NoError(t, bs.SetMode(mode.ReadWrite)) require.NoError(t, mb.SetMode(mode.ReadWrite)) wc.(*cache).flushed.Add(objects[0].addr.EncodeToString(), true) wc.(*cache).flushed.Add(objects[1].addr.EncodeToString(), false) require.NoError(t, wc.SetMode(mode.Degraded)) for i := 0; i < 2; i++ { var mPrm meta.GetPrm mPrm.SetAddress(objects[i].addr) _, err := mb.Get(mPrm) require.Error(t, err) _, err = bs.Get(common.GetPrm{Address: objects[i].addr}) require.Error(t, err) } check(t, mb, bs, objects[2:]) }) t.Run("ignore errors", func(t *testing.T) { testIgnoreErrors := func(t *testing.T, f func(*cache)) { var errCount atomic.Uint32 wc, bs, mb := newCache(t, WithReportErrorFunc(func(message string, err error) { errCount.Inc() })) objects := putObjects(t, wc) f(wc.(*cache)) require.NoError(t, wc.SetMode(mode.ReadOnly)) require.NoError(t, bs.SetMode(mode.ReadWrite)) require.NoError(t, mb.SetMode(mode.ReadWrite)) require.Equal(t, uint32(0), errCount.Load()) require.Error(t, wc.Flush(false)) require.True(t, errCount.Load() > 0) require.NoError(t, wc.Flush(true)) check(t, mb, bs, objects) } t.Run("db, invalid address", func(t *testing.T) { testIgnoreErrors(t, func(c *cache) { _, data := newObject(t, 1) require.NoError(t, c.db.Batch(func(tx *bbolt.Tx) error { b := tx.Bucket(defaultBucket) return b.Put([]byte{1, 2, 3}, data) })) }) }) t.Run("db, invalid object", func(t *testing.T) { testIgnoreErrors(t, func(c *cache) { require.NoError(t, c.db.Batch(func(tx *bbolt.Tx) error { b := tx.Bucket(defaultBucket) return b.Put([]byte(oidtest.Address().EncodeToString()), []byte{1, 2, 3}) })) }) }) t.Run("fs, read error", func(t *testing.T) { testIgnoreErrors(t, func(c *cache) { obj, data := newObject(t, 1) var prm common.PutPrm prm.Address = objectCore.AddressOf(obj) prm.RawData = data _, err := c.fsTree.Put(prm) require.NoError(t, err) p := prm.Address.Object().EncodeToString() + "." + prm.Address.Container().EncodeToString() p = filepath.Join(c.fsTree.RootPath, p[:1], p[1:]) _, err = os.Stat(p) // sanity check require.NoError(t, err) require.NoError(t, os.Truncate(p, 0)) // corrupt the file contents, so that it can't be unmarshalled }) }) t.Run("fs, invalid object", func(t *testing.T) { testIgnoreErrors(t, func(c *cache) { var prm common.PutPrm prm.Address = oidtest.Address() prm.RawData = []byte{1, 2, 3} _, err := c.fsTree.Put(prm) require.NoError(t, err) }) }) }) t.Run("on init", func(t *testing.T) { wc, bs, mb := newCache(t) objects := []objectPair{ // removed putObject(t, wc, 1), putObject(t, wc, smallSize+1), // not found putObject(t, wc, 1), putObject(t, wc, smallSize+1), // ok putObject(t, wc, 1), putObject(t, wc, smallSize+1), } require.NoError(t, wc.Close()) require.NoError(t, bs.SetMode(mode.ReadWrite)) require.NoError(t, mb.SetMode(mode.ReadWrite)) for i := range objects { var prm meta.PutPrm prm.SetObject(objects[i].obj) _, err := mb.Put(prm) require.NoError(t, err) } var inhumePrm meta.InhumePrm inhumePrm.SetAddresses(objects[0].addr, objects[1].addr) inhumePrm.SetTombstoneAddress(oidtest.Address()) _, err := mb.Inhume(inhumePrm) require.NoError(t, err) var deletePrm meta.DeletePrm deletePrm.SetAddresses(objects[2].addr, objects[3].addr) _, err = mb.Delete(deletePrm) require.NoError(t, err) require.NoError(t, bs.SetMode(mode.ReadOnly)) require.NoError(t, mb.SetMode(mode.ReadOnly)) // Open in read-only: no error, nothing is removed. require.NoError(t, wc.Open(true)) initWC(t, wc) for i := range objects { _, err := wc.Get(objects[i].addr) require.NoError(t, err, i) } require.NoError(t, wc.Close()) // Open in read-write: no error, something is removed. require.NoError(t, wc.Open(false)) initWC(t, wc) for i := range objects { _, err := wc.Get(objects[i].addr) if i < 2 { require.ErrorAs(t, err, new(apistatus.ObjectNotFound), i) } else { require.NoError(t, err, i) } } }) } func putObject(t *testing.T, c Cache, size int) objectPair { obj, data := newObject(t, size) var prm common.PutPrm prm.Address = objectCore.AddressOf(obj) prm.Object = obj prm.RawData = data _, err := c.Put(prm) require.NoError(t, err) return objectPair{prm.Address, prm.Object} } func newObject(t *testing.T, size int) (*object.Object, []byte) { obj := object.New() ver := versionSDK.Current() obj.SetID(oidtest.ID()) obj.SetOwnerID(usertest.ID()) obj.SetContainerID(cidtest.ID()) obj.SetType(object.TypeRegular) obj.SetVersion(&ver) obj.SetPayloadChecksum(checksumtest.Checksum()) obj.SetPayloadHomomorphicHash(checksumtest.Checksum()) obj.SetPayload(make([]byte, size)) data, err := obj.Marshal() require.NoError(t, err) return obj, data } func initWC(t *testing.T, wc Cache) { require.NoError(t, wc.Init()) require.Eventually(t, func() bool { rawWc := wc.(*cache) return rawWc.initialized.Load() }, 100*time.Second, 1*time.Millisecond) } type dummyEpoch struct{} func (dummyEpoch) CurrentEpoch() uint64 { return 0 }