frostfs-node/pkg/local_object_storage/shard/delete.go
Dmitrii Stepanov 1dd84eef77 [#384] shard: Cancel GC if change mode requested
Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
2023-05-25 09:55:16 +03:00

115 lines
3 KiB
Go

package shard
import (
"context"
"errors"
"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.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) {
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
}
for _, addr := range prm.addr {
select {
case <-ctx.Done():
return DeleteRes{}, ctx.Err()
default:
}
s.deleteObjectFromWriteCacheSafe(addr)
s.deleteFromBlobstorSafe(addr)
if err := s.deleteFromMetabase(addr); err != nil {
return DeleteRes{}, err // stop on metabase error ?
}
}
return DeleteRes{}, nil
}
func (s *Shard) deleteObjectFromWriteCacheSafe(addr oid.Address) {
if s.hasWriteCache() {
err := s.writeCache.Delete(addr)
if err != nil && !IsErrNotFound(err) && !errors.Is(err, writecache.ErrReadOnly) {
s.log.Warn("can't delete object from write cache", zap.String("error", err.Error()))
}
}
}
func (s *Shard) deleteFromMetabase(addr oid.Address) error {
var delPrm meta.DeletePrm
delPrm.SetAddresses(addr)
res, err := s.metaBase.Delete(delPrm)
if err != nil {
return err
}
s.decObjectCounterBy(physical, res.RawObjectsRemoved())
s.decObjectCounterBy(logical, res.AvailableObjectsRemoved())
removedPayload := res.RemovedPhysicalObjectSizes()[0]
logicalRemovedPayload := res.RemovedLogicalObjectSizes()[0]
if logicalRemovedPayload > 0 {
s.addToContainerSize(addr.Container().EncodeToString(), -int64(logicalRemovedPayload))
}
s.addToPayloadSize(-int64(removedPayload))
return nil
}
func (s *Shard) deleteFromBlobstorSafe(addr oid.Address) {
var sPrm meta.StorageIDPrm
sPrm.SetAddress(addr)
res, err := s.metaBase.StorageID(sPrm)
if err != nil {
s.log.Debug("can't get storage ID from metabase",
zap.Stringer("object", addr),
zap.String("error", err.Error()))
}
storageID := res.StorageID()
var delPrm common.DeletePrm
delPrm.Address = addr
delPrm.StorageID = storageID
_, err = s.blobStor.Delete(delPrm)
if err != nil {
s.log.Debug("can't remove object from blobStor",
zap.Stringer("object_address", addr),
zap.String("error", err.Error()))
}
}