[#1461] node: Allow force LOCK removal

Signed-off-by: Pavel Karpy <carpawell@nspcc.ru>
This commit is contained in:
Pavel Karpy 2022-05-31 23:11:42 +03:00 committed by fyrchik
parent 25826bd96d
commit 9c5ef3bab8
4 changed files with 64 additions and 2 deletions

View file

@ -36,6 +36,7 @@ func initControlService(c *cfg) {
controlSvc.WithDeletedObjectHandler(func(addrList []oid.Address) error { controlSvc.WithDeletedObjectHandler(func(addrList []oid.Address) error {
var prm engine.DeletePrm var prm engine.DeletePrm
prm.WithAddresses(addrList...) prm.WithAddresses(addrList...)
prm.WithForceRemoval()
_, err := c.cfgObject.cfgLocalStorage.localStorage.Delete(prm) _, err := c.cfgObject.cfgLocalStorage.localStorage.Delete(prm)

View file

@ -11,6 +11,8 @@ import (
// DeletePrm groups the parameters of Delete operation. // DeletePrm groups the parameters of Delete operation.
type DeletePrm struct { type DeletePrm struct {
addr []oid.Address addr []oid.Address
forceRemoval bool
} }
// DeleteRes groups the resulting values of Delete operation. // DeleteRes groups the resulting values of Delete operation.
@ -25,6 +27,15 @@ func (p *DeletePrm) WithAddresses(addr ...oid.Address) {
} }
} }
// WithForceRemoval is a Delete option to remove an object despite any
// restrictions imposed on deleting that object. Expected to be used
// only in control service.
func (p *DeletePrm) WithForceRemoval() {
if p != nil {
p.forceRemoval = true
}
}
// Delete marks the objects to be removed. // Delete marks the objects to be removed.
// //
// Returns an error if executions are blocked (see BlockExecution). // Returns an error if executions are blocked (see BlockExecution).
@ -65,6 +76,9 @@ func (e *StorageEngine) delete(prm DeletePrm) (DeleteRes, error) {
var shPrm shard.InhumePrm var shPrm shard.InhumePrm
shPrm.MarkAsGarbage(prm.addr[i]) shPrm.MarkAsGarbage(prm.addr[i])
if prm.forceRemoval {
shPrm.ForceRemoval()
}
_, err = sh.Inhume(shPrm) _, err = sh.Inhume(shPrm)
if err != nil { if err != nil {

View file

@ -18,6 +18,8 @@ type InhumePrm struct {
target []oid.Address target []oid.Address
lockObjectHandling bool lockObjectHandling bool
forceRemoval bool
} }
// InhumeRes encapsulates results of Inhume operation. // InhumeRes encapsulates results of Inhume operation.
@ -55,6 +57,7 @@ func (p *InhumePrm) WithTombstoneAddress(addr oid.Address) {
func (p *InhumePrm) WithGCMark() { func (p *InhumePrm) WithGCMark() {
if p != nil { if p != nil {
p.tomb = nil p.tomb = nil
p.forceRemoval = false
} }
} }
@ -66,6 +69,15 @@ func (p *InhumePrm) WithLockObjectHandling() {
} }
} }
// WithForceGCMark allows removal any object. Expected to be
// called only in control service.
func (p *InhumePrm) WithForceGCMark() {
if p != nil {
p.tomb = nil
p.forceRemoval = true
}
}
// Inhume inhumes the object by specified address. // Inhume inhumes the object by specified address.
// //
// tomb should not be nil. // tomb should not be nil.
@ -130,6 +142,19 @@ func (db *DB) Inhume(prm InhumePrm) (res InhumeRes, err error) {
return apistatus.ObjectLocked{} return apistatus.ObjectLocked{}
} }
var lockWasChecked bool
// prevent lock objects to be inhumed
// if `Inhume` was called not with the
// `WithForceGCMark` option
if !prm.forceRemoval {
if isLockObject(tx, cnr, id) {
return fmt.Errorf("lock object removal, CID: %s, OID: %s", cnr, id)
}
lockWasChecked = true
}
obj, err := db.get(tx, prm.target[i], false, true) obj, err := db.get(tx, prm.target[i], false, true)
// if object is stored and it is regular object then update bucket // if object is stored and it is regular object then update bucket
@ -184,6 +209,14 @@ func (db *DB) Inhume(prm InhumePrm) (res InhumeRes, err error) {
} }
if prm.lockObjectHandling { if prm.lockObjectHandling {
// do not perform lock check if
// it was already called
if lockWasChecked {
// inhumed object is not of
// the LOCK type
continue
}
if isLockObject(tx, cnr, id) { if isLockObject(tx, cnr, id) {
res.deletedLockObj = append(res.deletedLockObj, prm.target[i]) res.deletedLockObj = append(res.deletedLockObj, prm.target[i])
} }

View file

@ -11,8 +11,9 @@ import (
// InhumePrm encapsulates parameters for inhume operation. // InhumePrm encapsulates parameters for inhume operation.
type InhumePrm struct { type InhumePrm struct {
target []oid.Address target []oid.Address
tombstone *oid.Address tombstone *oid.Address
forceRemoval bool
} }
// InhumeRes encapsulates results of inhume operation. // InhumeRes encapsulates results of inhume operation.
@ -40,6 +41,15 @@ func (p *InhumePrm) MarkAsGarbage(addr ...oid.Address) {
} }
} }
// ForceRemoval forces object removing despite any restrictions imposed
// on deleting that object. Expected to be used only in control service.
func (p *InhumePrm) ForceRemoval() {
if p != nil {
p.tombstone = nil
p.forceRemoval = true
}
}
// Inhume calls metabase. Inhume method to mark object as removed. It won't be // Inhume calls metabase. Inhume method to mark object as removed. It won't be
// removed physically from blobStor and metabase until `Delete` operation. // removed physically from blobStor and metabase until `Delete` operation.
// //
@ -68,6 +78,10 @@ func (s *Shard) Inhume(prm InhumePrm) (InhumeRes, error) {
metaPrm.WithGCMark() metaPrm.WithGCMark()
} }
if prm.forceRemoval {
metaPrm.WithForceGCMark()
}
res, err := s.metaBase.Inhume(metaPrm) res, err := s.metaBase.Inhume(metaPrm)
if err != nil { if err != nil {
s.log.Debug("could not mark object to delete in metabase", s.log.Debug("could not mark object to delete in metabase",