forked from TrueCloudLab/frostfs-node
130 lines
3.4 KiB
Go
130 lines
3.4 KiB
Go
package shard
|
|
|
|
import (
|
|
"context"
|
|
|
|
"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"
|
|
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.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) 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 {
|
|
s.log.Debug(logs.ObjectRemovalFailureBlobStor,
|
|
zap.Stringer("object_address", addr),
|
|
zap.String("error", err.Error()),
|
|
zap.String("trace_id", tracingPkg.GetTraceID(ctx)))
|
|
}
|
|
return err
|
|
}
|
|
|
|
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.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
|
|
}
|