frostfs-node/pkg/local_object_storage/shard/delete.go
Dmitrii Stepanov 3b7c0362a8
All checks were successful
DCO action / DCO (pull_request) Successful in 12m22s
Build / Build Components (1.20) (pull_request) Successful in 12m44s
Build / Build Components (1.21) (pull_request) Successful in 13m2s
Vulncheck / Vulncheck (pull_request) Successful in 13m3s
Tests and linters / Staticcheck (pull_request) Successful in 15m32s
Tests and linters / Lint (pull_request) Successful in 16m26s
Tests and linters / Tests (1.21) (pull_request) Successful in 26m11s
Tests and linters / Tests with -race (pull_request) Successful in 28m29s
Tests and linters / Tests (1.20) (pull_request) Successful in 3m28s
[#861] shard: Fix Delete object
It is possible that object doesn't exist in metabase.

Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
2023-12-12 14:25:40 +03:00

156 lines
4.1 KiB
Go

package shard
import (
"context"
"fmt"
"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"
tracingPkg "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/tracing"
"git.frostfs.info/TrueCloudLab/frostfs-observability/tracing"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client"
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 {
deleted uint64
}
// 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 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, false)
}
func (s *Shard) delete(ctx context.Context, prm DeletePrm, skipFailed bool) (DeleteRes, error) {
if s.info.Mode.ReadOnly() {
return DeleteRes{}, ErrReadOnlyMode
} else if s.info.Mode.NoMetabase() {
return DeleteRes{}, ErrDegradedMode
}
result := DeleteRes{}
for _, addr := range prm.addr {
select {
case <-ctx.Done():
return result, ctx.Err()
default:
}
if err := s.validateWritecacheDoesntContainObject(ctx, addr); err != nil {
if skipFailed {
continue
}
return result, err
}
if err := s.deleteFromBlobstor(ctx, addr); err != nil {
if skipFailed {
continue
}
return result, err
}
if err := s.deleteFromMetabase(ctx, addr); err != nil {
if skipFailed {
continue
}
return result, err
}
result.deleted++
}
return result, nil
}
func (s *Shard) validateWritecacheDoesntContainObject(ctx context.Context, addr oid.Address) error {
if !s.hasWriteCache() {
return nil
}
_, err := s.writeCache.Head(ctx, addr)
if err == nil {
s.log.Warn(logs.ObjectRemovalFailureExistsInWritecache, zap.Stringer("object_address", addr))
return fmt.Errorf("object %s must be flushed from writecache", addr)
}
if client.IsErrObjectNotFound(err) {
return nil
}
return err
}
func (s *Shard) deleteFromBlobstor(ctx context.Context, addr oid.Address) error {
var sPrm meta.StorageIDPrm
sPrm.SetAddress(addr)
res, err := s.metaBase.StorageID(ctx, sPrm)
if err != nil {
s.log.Debug(logs.StorageIDRetrievalFailure,
zap.Stringer("object", addr),
zap.String("error", err.Error()),
zap.String("trace_id", tracingPkg.GetTraceID(ctx)))
return err
}
storageID := res.StorageID()
var delPrm common.DeletePrm
delPrm.Address = addr
delPrm.StorageID = storageID
_, err = s.blobStor.Delete(ctx, delPrm)
if err != nil && !client.IsErrObjectNotFound(err) {
s.log.Debug(logs.ObjectRemovalFailureBlobStor,
zap.Stringer("object_address", addr),
zap.String("error", err.Error()),
zap.String("trace_id", tracingPkg.GetTraceID(ctx)))
return err
}
return nil
}
func (s *Shard) deleteFromMetabase(ctx context.Context, addr oid.Address) error {
var delPrm meta.DeletePrm
delPrm.SetAddresses(addr)
res, err := s.metaBase.Delete(ctx, delPrm)
if err != nil {
return err
}
s.decObjectCounterBy(physical, res.RawObjectsRemoved())
s.decObjectCounterBy(logical, res.AvailableObjectsRemoved())
s.decObjectCounterBy(user, res.UserObjectsRemoved())
s.decContainerObjectCounter(res.RemovedByCnrID())
removedPayload := res.RemovedPhysicalObjectSizes()[0]
logicalRemovedPayload := res.RemovedLogicalObjectSizes()[0]
if logicalRemovedPayload > 0 {
s.addToContainerSize(addr.Container().EncodeToString(), -int64(logicalRemovedPayload))
}
s.addToPayloadSize(-int64(removedPayload))
return nil
}