[#574] core: Extend Source interface with DeletionInfo method

* Introduce common method EverExisted
* Define DeletionInfo for struct that must implement Source
* Refactor tree srv

Signed-off-by: Airat Arifullin <a.arifullin@yadro.com>
traceid
Airat Arifullin 2023-08-24 15:27:24 +03:00 committed by Evgenii Stratonikov
parent 9072772a09
commit 554ff2c06b
8 changed files with 115 additions and 43 deletions

View File

@ -142,7 +142,8 @@ func (c *lruNetCache) get(key uint64) (*netmapSDK.NetMap, error) {
// wrapper over TTL cache of values read from the network // wrapper over TTL cache of values read from the network
// that implements container storage. // that implements container storage.
type ttlContainerStorage struct { type ttlContainerStorage struct {
*ttlNetCache[cid.ID, *container.Container] containerCache *ttlNetCache[cid.ID, *container.Container]
delInfoCache *ttlNetCache[cid.ID, *container.DelInfo]
} }
func newCachedContainerStorage(v container.Source, ttl time.Duration) ttlContainerStorage { func newCachedContainerStorage(v container.Source, ttl time.Duration) ttlContainerStorage {
@ -151,18 +152,32 @@ func newCachedContainerStorage(v container.Source, ttl time.Duration) ttlContain
lruCnrCache := newNetworkTTLCache(containerCacheSize, ttl, func(id cid.ID) (*container.Container, error) { lruCnrCache := newNetworkTTLCache(containerCacheSize, ttl, func(id cid.ID) (*container.Container, error) {
return v.Get(id) return v.Get(id)
}) })
lruDelInfoCache := newNetworkTTLCache(containerCacheSize, ttl, func(id cid.ID) (*container.DelInfo, error) {
return v.DeletionInfo(id)
})
return ttlContainerStorage{lruCnrCache} return ttlContainerStorage{
containerCache: lruCnrCache,
delInfoCache: lruDelInfoCache,
}
} }
func (s ttlContainerStorage) handleRemoval(cnr cid.ID) { func (s ttlContainerStorage) handleRemoval(cnr cid.ID) {
s.set(cnr, nil, new(apistatus.ContainerNotFound)) s.containerCache.set(cnr, nil, new(apistatus.ContainerNotFound))
// The removal causes the cache miss and thus deletion info (that contains
// ownerID and epoch) for the container will be updated from sidechain.
s.delInfoCache.remove(cnr)
} }
// Get returns container value from the cache. If value is missing in the cache // Get returns container value from the cache. If value is missing in the cache
// or expired, then it returns value from side chain and updates the cache. // or expired, then it returns value from side chain and updates the cache.
func (s ttlContainerStorage) Get(cnr cid.ID) (*container.Container, error) { func (s ttlContainerStorage) Get(cnr cid.ID) (*container.Container, error) {
return s.get(cnr) return s.containerCache.get(cnr)
}
func (s ttlContainerStorage) DeletionInfo(cnr cid.ID) (*container.DelInfo, error) {
return s.delInfoCache.get(cnr)
} }
type ttlEACLStorage struct { type ttlEACLStorage struct {

View File

@ -28,6 +28,7 @@ import (
containerMorph "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/container/morph" containerMorph "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/container/morph"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger" "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger"
apiClient "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client" apiClient "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client"
apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
containerSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container" containerSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container"
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id" cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap" "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap"
@ -113,7 +114,7 @@ func configureEACLAndContainerSources(c *cfg, client *cntClient.Client, cnrSrc c
c.cfgObject.eaclSource = eACLFetcher c.cfgObject.eaclSource = eACLFetcher
cnrRdr.eacl = eACLFetcher cnrRdr.eacl = eACLFetcher
c.cfgObject.cnrSource = cnrSrc c.cfgObject.cnrSource = cnrSrc
cnrRdr.get = cnrSrc cnrRdr.src = cnrSrc
cnrRdr.lister = client cnrRdr.lister = client
} else { } else {
// use RPC node as source of Container contract items (with caching) // use RPC node as source of Container contract items (with caching)
@ -131,7 +132,8 @@ func configureEACLAndContainerSources(c *cfg, client *cntClient.Client, cnrSrc c
cnr, err := cnrSrc.Get(ev.ID) cnr, err := cnrSrc.Get(ev.ID)
if err == nil { if err == nil {
cachedContainerLister.update(cnr.Value.Owner(), ev.ID, true) cachedContainerLister.update(cnr.Value.Owner(), ev.ID, true)
cachedContainerStorage.set(ev.ID, cnr, nil) cachedContainerStorage.containerCache.set(ev.ID, cnr, nil)
cachedContainerStorage.delInfoCache.set(ev.ID, nil, new(apistatus.ContainerNotFound))
} else { } else {
// unlike removal, we expect successful receive of the container // unlike removal, we expect successful receive of the container
// after successful creation, so logging can be useful // after successful creation, so logging can be useful
@ -159,7 +161,6 @@ func configureEACLAndContainerSources(c *cfg, client *cntClient.Client, cnrSrc c
} }
cachedContainerStorage.handleRemoval(ev.ID) cachedContainerStorage.handleRemoval(ev.ID)
c.log.Debug(logs.FrostFSNodeContainerRemovalEventsReceipt, c.log.Debug(logs.FrostFSNodeContainerRemovalEventsReceipt,
zap.Stringer("id", ev.ID), zap.Stringer("id", ev.ID),
) )
@ -170,7 +171,7 @@ func configureEACLAndContainerSources(c *cfg, client *cntClient.Client, cnrSrc c
cnrRdr.lister = cachedContainerLister cnrRdr.lister = cachedContainerLister
cnrRdr.eacl = c.cfgObject.eaclSource cnrRdr.eacl = c.cfgObject.eaclSource
cnrRdr.get = c.cfgObject.cnrSource cnrRdr.src = c.cfgObject.cnrSource
cnrWrt.cacheEnabled = true cnrWrt.cacheEnabled = true
cnrWrt.eacls = cachedEACLStorage cnrWrt.eacls = cachedEACLStorage
@ -641,7 +642,7 @@ func (c *usedSpaceService) processLoadValue(_ context.Context, a containerSDK.Si
type morphContainerReader struct { type morphContainerReader struct {
eacl containerCore.EACLSource eacl containerCore.EACLSource
get containerCore.Source src containerCore.Source
lister interface { lister interface {
List(*user.ID) ([]cid.ID, error) List(*user.ID) ([]cid.ID, error)
@ -649,7 +650,11 @@ type morphContainerReader struct {
} }
func (x *morphContainerReader) Get(id cid.ID) (*containerCore.Container, error) { func (x *morphContainerReader) Get(id cid.ID) (*containerCore.Container, error) {
return x.get.Get(id) return x.src.Get(id)
}
func (x *morphContainerReader) DeletionInfo(id cid.ID) (*containerCore.DelInfo, error) {
return x.src.DeletionInfo(id)
} }
func (x *morphContainerReader) GetEACL(id cid.ID) (*containerCore.EACL, error) { func (x *morphContainerReader) GetEACL(id cid.ID) (*containerCore.EACL, error) {

View File

@ -32,7 +32,7 @@ func (c cnrSource) Get(id cid.ID) (*container.Container, error) {
} }
func (c cnrSource) DeletionInfo(cid cid.ID) (*container.DelInfo, error) { func (c cnrSource) DeletionInfo(cid cid.ID) (*container.DelInfo, error) {
return c.cli.DeletionInfo(cid) return c.src.DeletionInfo(cid)
} }
func (c cnrSource) List() ([]cid.ID, error) { func (c cnrSource) List() ([]cid.ID, error) {

View File

@ -41,6 +41,8 @@ type Source interface {
// Implementations must not retain the container pointer and modify // Implementations must not retain the container pointer and modify
// the container through it. // the container through it.
Get(cid.ID) (*Container, error) Get(cid.ID) (*Container, error)
DeletionInfo(cid.ID) (*DelInfo, error)
} }
// EACL groups information about the FrostFS container's extended ACL stored in // EACL groups information about the FrostFS container's extended ACL stored in

View File

@ -0,0 +1,22 @@
package container
import (
"errors"
apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
)
// EverExisted checks whether the container ever existed or
// it just has not been created yet at the current epoch.
func EverExisted(s Source, cid cid.ID) (bool, error) {
_, err := s.DeletionInfo(cid)
if err == nil {
return true, nil
}
var errContainerNotFound *apistatus.ContainerNotFound
if errors.As(err, &errContainerNotFound) {
return false, nil
}
return false, err
}

View File

@ -1,6 +1,7 @@
package container package container
import ( import (
"crypto/sha256"
"fmt" "fmt"
"strings" "strings"
@ -11,7 +12,26 @@ import (
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id" cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
) )
func (c *Client) DeletionInfo(cid cid.ID) (*containercore.DelInfo, error) { func (x *containerSource) DeletionInfo(cnr cid.ID) (*containercore.DelInfo, error) {
return DeletionInfo((*Client)(x), cnr)
}
type deletionInfo interface {
DeletionInfo(cid []byte) (*containercore.DelInfo, error)
}
func AsContainerSpecInfoProvider(w *Client) containercore.Source {
return (*containerSource)(w)
}
func DeletionInfo(c deletionInfo, cnr cid.ID) (*containercore.DelInfo, error) {
binCnr := make([]byte, sha256.Size)
cnr.Encode(binCnr)
return c.DeletionInfo(binCnr)
}
func (c *Client) DeletionInfo(cid []byte) (*containercore.DelInfo, error) {
prm := client.TestInvokePrm{} prm := client.TestInvokePrm{}
prm.SetMethod(deletionInfoMethod) prm.SetMethod(deletionInfoMethod)
prm.SetArgs(cid) prm.SetArgs(cid)

View File

@ -34,8 +34,13 @@ func TestBuryObjectWithoutContainer(t *testing.T) {
// Container source and bury function // Container source and bury function
buryCh := make(chan oid.Address) buryCh := make(chan oid.Address)
containerSrc := func(id cid.ID) (*container.Container, error) { containerSrc := containerSrc{
return nil, new(apistatus.ContainerNotFound) get: func(id cid.ID) (*container.Container, error) {
return nil, new(apistatus.ContainerNotFound)
},
deletionInfo: func(id cid.ID) (*container.DelInfo, error) {
return &container.DelInfo{}, nil
},
} }
buryFn := func(ctx context.Context, a oid.Address) error { buryFn := func(ctx context.Context, a oid.Address) error {
buryCh <- a buryCh <- a
@ -49,7 +54,7 @@ func TestBuryObjectWithoutContainer(t *testing.T) {
// Policer instance // Policer instance
p := New( p := New(
WithKeySpaceIterator(&sliceKeySpaceIterator{objs: objs}), WithKeySpaceIterator(&sliceKeySpaceIterator{objs: objs}),
WithContainerSource(containerSrcFunc(containerSrc)), WithContainerSource(containerSrc),
WithBuryFunc(buryFn), WithBuryFunc(buryFn),
WithPool(pool), WithPool(pool),
) )
@ -194,12 +199,17 @@ func TestProcessObject(t *testing.T) {
cnr := &container.Container{} cnr := &container.Container{}
cnr.Value.Init() cnr.Value.Init()
cnr.Value.SetPlacementPolicy(policy) cnr.Value.SetPlacementPolicy(policy)
containerSrc := func(id cid.ID) (*container.Container, error) { containerSrc := containerSrc{
if id.Equals(addr.Container()) { get: func(id cid.ID) (*container.Container, error) {
return cnr, nil if id.Equals(addr.Container()) {
} return cnr, nil
t.Errorf("unexpected container requested: got=%v, want=%v", id, addr.Container()) }
return nil, new(apistatus.ContainerNotFound) t.Errorf("unexpected container requested: got=%v, want=%v", id, addr.Container())
return nil, new(apistatus.ContainerNotFound)
},
deletionInfo: func(id cid.ID) (*container.DelInfo, error) {
return &container.DelInfo{}, nil
},
} }
buryFn := func(ctx context.Context, a oid.Address) error { buryFn := func(ctx context.Context, a oid.Address) error {
t.Errorf("unexpected object buried: %v", a) t.Errorf("unexpected object buried: %v", a)
@ -211,7 +221,7 @@ func TestProcessObject(t *testing.T) {
var gotReplicateTo []int var gotReplicateTo []int
p := New( p := New(
WithContainerSource(containerSrcFunc(containerSrc)), WithContainerSource(containerSrc),
WithPlacementBuilder(placementBuilderFunc(placementBuilder)), WithPlacementBuilder(placementBuilderFunc(placementBuilder)),
WithNetmapKeys(announcedKeysFunc(func(k []byte) bool { WithNetmapKeys(announcedKeysFunc(func(k []byte) bool {
return bytes.Equal(k, nodes[0].PublicKey()) return bytes.Equal(k, nodes[0].PublicKey())
@ -251,9 +261,6 @@ func TestIteratorContract(t *testing.T) {
Type: objectSDK.TypeRegular, Type: objectSDK.TypeRegular,
}} }}
containerSrc := func(id cid.ID) (*container.Container, error) {
return nil, new(apistatus.ContainerNotFound)
}
buryFn := func(ctx context.Context, a oid.Address) error { buryFn := func(ctx context.Context, a oid.Address) error {
return nil return nil
} }
@ -273,9 +280,18 @@ func TestIteratorContract(t *testing.T) {
finishCh: make(chan struct{}), finishCh: make(chan struct{}),
} }
containerSrc := containerSrc{
get: func(id cid.ID) (*container.Container, error) {
return nil, new(apistatus.ContainerNotFound)
},
deletionInfo: func(id cid.ID) (*container.DelInfo, error) {
return &container.DelInfo{}, nil
},
}
p := New( p := New(
WithKeySpaceIterator(it), WithKeySpaceIterator(it),
WithContainerSource(containerSrcFunc(containerSrc)), WithContainerSource(containerSrc),
WithBuryFunc(buryFn), WithBuryFunc(buryFn),
WithPool(pool), WithPool(pool),
func(c *cfg) { func(c *cfg) {
@ -353,10 +369,14 @@ func (it *sliceKeySpaceIterator) Rewind() {
it.cur = 0 it.cur = 0
} }
// containerSrcFunc is a container.Source backed by a function. type containerSrc struct {
type containerSrcFunc func(cid.ID) (*container.Container, error) get func(id cid.ID) (*container.Container, error)
deletionInfo func(id cid.ID) (*container.DelInfo, error)
}
func (f containerSrcFunc) Get(id cid.ID) (*container.Container, error) { return f(id) } func (f containerSrc) Get(id cid.ID) (*container.Container, error) { return f.get(id) }
func (f containerSrc) DeletionInfo(id cid.ID) (*container.DelInfo, error) { return f.deletionInfo(id) }
// placementBuilderFunc is a placement.Builder backed by a function // placementBuilderFunc is a placement.Builder backed by a function
type placementBuilderFunc func(cid.ID, *oid.ID, netmap.PlacementPolicy) ([][]netmap.NodeInfo, error) type placementBuilderFunc func(cid.ID, *oid.ID, netmap.PlacementPolicy) ([][]netmap.NodeInfo, error)

View File

@ -12,13 +12,13 @@ import (
"time" "time"
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs" "git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
containerCore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/container"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/pilorama" "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/pilorama"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/morph/client/netmap" "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/morph/client/netmap"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/network" "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/network"
metrics "git.frostfs.info/TrueCloudLab/frostfs-observability/metrics/grpc" metrics "git.frostfs.info/TrueCloudLab/frostfs-observability/metrics/grpc"
tracing "git.frostfs.info/TrueCloudLab/frostfs-observability/tracing" tracing "git.frostfs.info/TrueCloudLab/frostfs-observability/tracing"
tracing_grpc "git.frostfs.info/TrueCloudLab/frostfs-observability/tracing/grpc" tracing_grpc "git.frostfs.info/TrueCloudLab/frostfs-observability/tracing/grpc"
apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id" cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
netmapSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap" netmapSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap"
"github.com/panjf2000/ants/v2" "github.com/panjf2000/ants/v2"
@ -441,18 +441,6 @@ func (s *Service) syncContainers(ctx context.Context, cnrs []cid.ID) {
wg.Wait() wg.Wait()
} }
func (s *Service) containerEverExisted(cid cid.ID) (bool, error) {
_, err := s.cnrSource.DeletionInfo(cid)
if err == nil {
return true, nil
}
var errContainerNotFound *apistatus.ContainerNotFound
if errors.As(err, &errContainerNotFound) {
return false, nil
}
return false, err
}
func (s *Service) removeContainers(ctx context.Context, newContainers map[cid.ID]struct{}) { func (s *Service) removeContainers(ctx context.Context, newContainers map[cid.ID]struct{}) {
ctx, span := tracing.StartSpanFromContext(ctx, "TreeService.removeContainers") ctx, span := tracing.StartSpanFromContext(ctx, "TreeService.removeContainers")
defer span.End() defer span.End()
@ -466,7 +454,7 @@ func (s *Service) removeContainers(ctx context.Context, newContainers map[cid.ID
continue continue
} }
existed, err := s.containerEverExisted(cnr) existed, err := containerCore.EverExisted(s.cnrSource, cnr)
if err != nil { if err != nil {
s.log.Error(logs.TreeCouldNotCheckIfContainerExisted, s.log.Error(logs.TreeCouldNotCheckIfContainerExisted,
zap.Stringer("cid", cnr), zap.Stringer("cid", cnr),