frostfs-node/pkg/local_object_storage/shard/delete.go

package shard

import (
	"context"
	"errors"

	"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/pkg/tracing"
	"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/common"
	meta "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/metabase"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/writecache"
	oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/trace"
	"go.uber.org/zap"
)

// DeletePrm groups the parameters of Delete operation.
type DeletePrm struct {
	addr []oid.Address
}

// DeleteRes groups the resulting values of Delete operation.
type DeleteRes struct{}

// SetAddresses is a Delete option to set the addresses of the objects to delete.
//
// Option is required.
func (p *DeletePrm) SetAddresses(addr ...oid.Address) {
	p.addr = append(p.addr, addr...)
}

// Delete removes data from the shard's writeCache, metaBase and
// blobStor.
func (s *Shard) Delete(ctx context.Context, prm DeletePrm) (DeleteRes, error) {
	ctx, span := tracing.StartSpanFromContext(ctx, "Shard.Delete",
		trace.WithAttributes(
			attribute.String("shard_id", s.ID().String()),
			attribute.Int("addr_count", len(prm.addr)),
		))
	defer span.End()

	s.m.RLock()
	defer s.m.RUnlock()

	return s.delete(ctx, prm)
}

func (s *Shard) delete(ctx context.Context, prm DeletePrm) (DeleteRes, error) {
	if s.info.Mode.ReadOnly() {
		return DeleteRes{}, ErrReadOnlyMode
	} else if s.info.Mode.NoMetabase() {
		return DeleteRes{}, ErrDegradedMode
	}

	ln := len(prm.addr)

	smalls := make(map[oid.Address][]byte, ln)

	for i := range prm.addr {
		if s.hasWriteCache() {
			err := s.writeCache.Delete(ctx, prm.addr[i])
			if err != nil && !IsErrNotFound(err) && !errors.Is(err, writecache.ErrReadOnly) {
				s.log.Warn(logs.ShardCantDeleteObjectFromWriteCache, zap.String("error", err.Error()))
			}
		}

		var sPrm meta.StorageIDPrm
		sPrm.SetAddress(prm.addr[i])

		res, err := s.metaBase.StorageID(ctx, sPrm)
		if err != nil {
			s.log.Debug(logs.ShardCantGetStorageIDFromMetabase,
				zap.Stringer("object", prm.addr[i]),
				zap.String("error", err.Error()))

			continue
		}

		if res.StorageID() != nil {
			smalls[prm.addr[i]] = res.StorageID()
		}
	}

	var delPrm meta.DeletePrm
	delPrm.SetAddresses(prm.addr...)

	res, err := s.metaBase.Delete(ctx, delPrm)
	if err != nil {
		return DeleteRes{}, err // stop on metabase error ?
	}

	var totalRemovedPayload uint64

	s.decObjectCounterBy(physical, res.RawObjectsRemoved())
	s.decObjectCounterBy(logical, res.AvailableObjectsRemoved())
	for i := range prm.addr {
		removedPayload := res.RemovedPhysicalObjectSizes()[i]
		totalRemovedPayload += removedPayload
		logicalRemovedPayload := res.RemovedLogicalObjectSizes()[i]
		if logicalRemovedPayload > 0 {
			s.addToContainerSize(prm.addr[i].Container().EncodeToString(), -int64(logicalRemovedPayload))
		}
	}
	s.addToPayloadSize(-int64(totalRemovedPayload))

	for i := range prm.addr {
		var delPrm common.DeletePrm
		delPrm.Address = prm.addr[i]
		id := smalls[prm.addr[i]]
		delPrm.StorageID = id

		_, err = s.blobStor.Delete(ctx, delPrm)
		if err != nil {
			s.log.Debug(logs.ShardCantRemoveObjectFromBlobStor,
				zap.Stringer("object_address", prm.addr[i]),
				zap.String("error", err.Error()))
		}
	}

	return DeleteRes{}, nil
}