frostfs-node/pkg/local_object_storage/writecache/writecachebadger/flush.go

package writecachebadger

import (
	"bytes"
	"context"
	"encoding/hex"
	"errors"
	"fmt"
	"time"

	"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
	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/internal/metaerr"
	meta "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/metabase"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/writecache"
	"git.frostfs.info/TrueCloudLab/frostfs-observability/tracing"
	objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
	"github.com/dgraph-io/badger/v4"
	"github.com/dgraph-io/ristretto/z"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/trace"
	"go.uber.org/zap"
)

const (
	// flushBatchSize is amount of keys which will be read from cache to be flushed
	// to the main storage. It is used to reduce contention between cache put
	// and cache persist.
	flushBatchSize = 512
	// defaultFlushWorkersCount is number of workers for putting objects in main storage.
	defaultFlushWorkersCount = 20
	// defaultFlushInterval is default time interval between successive flushes.
	defaultFlushInterval = time.Second
)

type collector struct {
	cache     *cache
	scheduled int
	processed int
	cancel    func()
}

func (c *collector) Send(buf *z.Buffer) error {
	list, err := badger.BufferToKVList(buf)
	if err != nil {
		return err
	}
	for _, kv := range list.Kv {
		select {
		case <-c.cache.closeCh:
			c.cancel()
			return nil
		default:
		}
		if kv.StreamDone {
			return nil
		}
		if c.scheduled >= flushBatchSize {
			c.cancel()
			return nil
		}
		if got, want := len(kv.Key), len(internalKey{}); got != want {
			c.cache.log.Debug(
				fmt.Sprintf("not expected db key len: got %d, want %d", got, want))
			continue
		}
		c.processed++
		obj := objectSDK.New()
		val := bytes.Clone(kv.Value)
		if err = obj.Unmarshal(val); err != nil {
			continue
		}
		addr := objectCore.AddressOf(obj)
		c.cache.scheduled4FlushMtx.RLock()
		_, ok := c.cache.scheduled4Flush[addr]
		c.cache.scheduled4FlushMtx.RUnlock()
		if ok {
			c.cache.log.Debug(logs.WritecacheBadgerObjAlreadyScheduled, zap.Stringer("obj", addr))
			continue
		}
		c.cache.scheduled4FlushMtx.Lock()
		c.cache.scheduled4Flush[addr] = struct{}{}
		c.cache.scheduled4FlushMtx.Unlock()
		c.scheduled++
		select {
		case c.cache.flushCh <- objectInfo{
			addr: addr,
			data: val,
			obj:  obj,
		}:
		case <-c.cache.closeCh:
			c.cancel()
			return nil
		}
	}
	return nil
}

// runFlushLoop starts background workers which periodically flush objects to the blobstor.
func (c *cache) runFlushLoop() {
	for i := 0; i < c.workersCount; i++ {
		c.wg.Add(1)
		go c.workerFlushSmall()
	}

	c.wg.Add(1)
	go func() {
		defer c.wg.Done()

		tt := time.NewTimer(defaultFlushInterval)
		defer tt.Stop()

		for {
			select {
			case <-tt.C:
				c.flushSmallObjects()
				tt.Reset(defaultFlushInterval)
			case <-c.closeCh:
				return
			}
		}
	}()
}

func (c *cache) flushSmallObjects() {
	for {
		select {
		case <-c.closeCh:
			return
		default:
		}
		c.modeMtx.RLock()
		if c.readOnly() {
			c.modeMtx.RUnlock()
			time.Sleep(time.Second)
			continue
		}

		// Using the db after Close will panic and badger won't wait for outstanding txs,
		// so we need to check manually.
		if c.db.IsClosed() {
			c.modeMtx.RUnlock()
			return
		}
		ctx, cancel := context.WithCancel(context.TODO())
		coll := collector{
			cache:  c,
			cancel: cancel,
		}
		stream := c.db.NewStream()
		// All calls to Send are done by a single goroutine
		stream.Send = coll.Send
		if err := stream.Orchestrate(ctx); err != nil {
			c.log.Debug(fmt.Sprintf(
				"error during flushing object from wc: %s", err))
		}
		c.modeMtx.RUnlock()
		if coll.scheduled == 0 {
			break
		}
		c.log.Debug(logs.WritecacheTriedToFlushItemsFromWritecache,
			zap.Int("scheduled", coll.scheduled), zap.Int("processed", coll.processed))
	}
}

func (c *cache) reportFlushError(msg string, addr string, err error) {
	if c.reportError != nil {
		c.reportError(msg, err)
	} else {
		c.log.Error(msg,
			zap.String("address", addr),
			zap.Error(err))
	}
}

// workerFlushSmall writes small objects to the main storage.
func (c *cache) workerFlushSmall() {
	defer c.wg.Done()

	var objInfo objectInfo
	for {
		// Give priority to direct put.
		select {
		case objInfo = <-c.flushCh:
		case <-c.closeCh:
			return
		}

		err := c.flushObject(context.TODO(), objInfo.obj, objInfo.data, writecache.StorageTypeDB)
		if err == nil {
			c.deleteFromDB([]internalKey{addr2key(objInfo.addr)})
		}
		c.scheduled4FlushMtx.Lock()
		delete(c.scheduled4Flush, objInfo.addr)
		c.scheduled4FlushMtx.Unlock()
	}
}

// flushObject is used to write object directly to the main storage.
func (c *cache) flushObject(ctx context.Context, obj *objectSDK.Object, data []byte, st writecache.StorageType) error {
	var err error

	defer func() {
		c.metrics.Flush(err == nil, st)
	}()

	addr := objectCore.AddressOf(obj)

	var prm common.PutPrm
	prm.Object = obj
	prm.RawData = data

	res, err := c.blobstor.Put(ctx, prm)
	if err != nil {
		if !errors.Is(err, common.ErrNoSpace) && !errors.Is(err, common.ErrReadOnly) &&
			!errors.Is(err, blobstor.ErrNoPlaceFound) {
			c.reportFlushError(logs.FrostFSNodeCantFlushObjectToBlobstor,
				addr.EncodeToString(), err)
		}
		return err
	}

	var updPrm meta.UpdateStorageIDPrm
	updPrm.SetAddress(addr)
	updPrm.SetStorageID(res.StorageID)

	_, err = c.metabase.UpdateStorageID(updPrm)
	if err != nil {
		c.reportFlushError(logs.FrostFSNodeCantUpdateObjectStorageID,
			addr.EncodeToString(), err)
	}
	return err
}

// Flush flushes all objects from the write-cache to the main storage.
// Write-cache must be in readonly mode to ensure correctness of an operation and
// to prevent interference with background flush workers.
func (c *cache) Flush(ctx context.Context, ignoreErrors bool) error {
	ctx, span := tracing.StartSpanFromContext(ctx, "writecache.Flush",
		trace.WithAttributes(
			attribute.Bool("ignore_errors", ignoreErrors),
		))
	defer span.End()

	c.modeMtx.RLock()
	defer c.modeMtx.RUnlock()

	return c.flush(ctx, ignoreErrors)
}

func (c *cache) flush(ctx context.Context, ignoreErrors bool) error {
	return c.db.View(func(tx *badger.Txn) error {
		it := tx.NewIterator(badger.DefaultIteratorOptions)
		defer it.Close()
		var key internalKey
		for it.Rewind(); it.Valid(); it.Next() {
			if got, want := int(it.Item().KeySize()), len(key); got != want {
				err := fmt.Errorf("invalid db key len: got %d, want %d", got, want)
				c.reportFlushError(logs.FrostFSNodeCantDecodeObjectAddressFromDB, hex.EncodeToString(it.Item().Key()), metaerr.Wrap(err))
				if ignoreErrors {
					continue
				}
				return err
			}
			if err := it.Item().Value(func(data []byte) error {
				var obj objectSDK.Object
				if err := obj.Unmarshal(data); err != nil {
					copy(key[:], it.Item().Key())
					c.reportFlushError(logs.FrostFSNodeCantUnmarshalObjectFromDB, key.address().EncodeToString(), metaerr.Wrap(err))
					if ignoreErrors {
						return nil
					}
					return err
				}

				return c.flushObject(ctx, &obj, data, writecache.StorageTypeDB)
			}); err != nil {
				return err
			}
		}
		return nil
	})
}