-
Notifications
You must be signed in to change notification settings - Fork 38
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
WIP: node: Serve new object replication service
NeoFS protocol has been recently extended with new object replication RPC `ObjectService.Replicate` separated from the general-purpose `ObjectService.Put` one. According to API of the new RPC, all physically stored objects are transmitted in one message. Also, replication request and response formats are much simpler than for other operations. Serve new RPC by the storage node app. Signed-off-by: Leonard Lyubich <[email protected]>
- Loading branch information
1 parent
0e4fe58
commit 02b9fb9
Showing
10 changed files
with
1,639 additions
and
10 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,199 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "errors" | ||
| "fmt" | ||
| "strconv" | ||
|
|
||
| "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/engine" | ||
| cid "github.com/nspcc-dev/neofs-sdk-go/container/id" | ||
| netmapsdk "github.com/nspcc-dev/neofs-sdk-go/netmap" | ||
| "github.com/nspcc-dev/neofs-sdk-go/object" | ||
| oid "github.com/nspcc-dev/neofs-sdk-go/object/id" | ||
| "go.uber.org/zap" | ||
| ) | ||
|
|
||
| // interface of [engine.StorageEngine] useful for testing. | ||
| type replicationNodeLocalObjectStorage interface { | ||
| IsLocked(oid.Address) (bool, error) | ||
| Put(engine.PutPrm) (engine.PutRes, error) | ||
| } | ||
|
|
||
| // replicationNode is [objectTransportGRPC.Node] interface provider | ||
| // checking storage policy compliance against NeoFS network maps. | ||
| type replicationNode struct { | ||
| log *zap.Logger | ||
|
|
||
| localObjStorage replicationNodeLocalObjectStorage | ||
|
|
||
| getPubKey func() []byte | ||
| getContainerStoragePolicy func(cid.ID) (netmapsdk.PlacementPolicy, error) | ||
| getCurrentEpoch func() uint64 | ||
| getNetmap func(epoch uint64) (netmapsdk.NetMap, error) | ||
| } | ||
|
|
||
| func newReplicationNode( | ||
| log *zap.Logger, | ||
| localObjStorage replicationNodeLocalObjectStorage, | ||
| getLocalNodePubKey func() []byte, | ||
| getContainerStoragePolicy func(cid.ID) (netmapsdk.PlacementPolicy, error), | ||
| getCurrentEpoch func() uint64, | ||
| getNetmap func(epoch uint64) (netmapsdk.NetMap, error), | ||
| ) *replicationNode { | ||
| return &replicationNode{ | ||
| log: log, | ||
| localObjStorage: localObjStorage, | ||
| getPubKey: getLocalNodePubKey, | ||
| getContainerStoragePolicy: getContainerStoragePolicy, | ||
| getCurrentEpoch: getCurrentEpoch, | ||
| getNetmap: getNetmap, | ||
| } | ||
| } | ||
|
|
||
| func (x *replicationNode) compliesStoragePolicyInPastNetmap(bPubKey []byte, cnrID cid.ID, policy netmapsdk.PlacementPolicy, epoch uint64) (bool, error) { | ||
| nm, err := x.getNetmap(epoch) | ||
| if err != nil { | ||
| return false, fmt.Errorf("read network map: %w", err) | ||
| } | ||
|
|
||
| inNetmap := false | ||
| nodes := nm.Nodes() | ||
|
|
||
| for i := range nodes { | ||
| if bytes.Equal(nodes[i].PublicKey(), bPubKey) { | ||
| inNetmap = true | ||
| break | ||
| } | ||
| } | ||
|
|
||
| if !inNetmap { | ||
| return false, nil | ||
| } | ||
|
|
||
| cnrVectors, err := nm.ContainerNodes(policy, cnrID) | ||
| if err != nil { | ||
| return false, fmt.Errorf("build list of container nodes from network map, storage policy and container ID: %w", err) | ||
| } | ||
|
|
||
| for i := range cnrVectors { | ||
| for j := range cnrVectors[i] { | ||
| if bytes.Equal(cnrVectors[i][j].PublicKey(), bPubKey) { | ||
| return true, nil | ||
| } | ||
| } | ||
| } | ||
|
|
||
| return false, nil | ||
| } | ||
|
|
||
| // CompliesContainerStoragePolicy checks whether local node's public key is | ||
| // presented in network map of the latest NeoFS epoch and matches storage policy | ||
| // of the referenced container. | ||
| func (x *replicationNode) CompliesContainerStoragePolicy(cnrID cid.ID) (bool, error) { | ||
| storagePolicy, err := x.getContainerStoragePolicy(cnrID) | ||
| if err != nil { | ||
| return false, fmt.Errorf("read container storage policy by container ID: %w", err) | ||
| } | ||
|
|
||
| ok, err := x.compliesStoragePolicyInPastNetmap(x.getPubKey(), cnrID, storagePolicy, x.getCurrentEpoch()) | ||
| if err != nil { | ||
| return false, fmt.Errorf("check with the latest network map: %w", err) | ||
| } | ||
|
|
||
| return ok, nil | ||
| } | ||
|
|
||
| // ClientCompliesContainerStoragePolicy checks whether given public key belongs | ||
| // to any storage node present in network map of the latest or previous NeoFS | ||
| // epoch and matching storage policy of the referenced container. | ||
| func (x *replicationNode) ClientCompliesContainerStoragePolicy(bClientPubKey []byte, cnrID cid.ID) (bool, error) { | ||
| storagePolicy, err := x.getContainerStoragePolicy(cnrID) | ||
| if err != nil { | ||
| return false, fmt.Errorf("read container storage policy by container ID: %w", err) | ||
| } | ||
|
|
||
| curEpoch := x.getCurrentEpoch() | ||
|
|
||
| ok, err := x.compliesStoragePolicyInPastNetmap(bClientPubKey, cnrID, storagePolicy, curEpoch) | ||
| if err != nil { | ||
| return false, fmt.Errorf("check with the latest network map: %w", err) | ||
| } | ||
|
|
||
| if !ok && curEpoch > 0 { | ||
| ok, err = x.compliesStoragePolicyInPastNetmap(bClientPubKey, cnrID, storagePolicy, curEpoch-1) | ||
| if err != nil { | ||
| return false, fmt.Errorf("check with previous network map: %w", err) | ||
| } | ||
| } | ||
|
|
||
| return ok, nil | ||
| } | ||
|
|
||
| var errObjectExpired = errors.New("object is expired") | ||
|
|
||
| // StoreBinaryObject checks whether binary-encoded NeoFS object from the | ||
| // referenced container is not expired (*) and, if it is, saves the object in | ||
| // the underlying [engine.StorageEngine]. | ||
| // | ||
| // (*) if the object is LOCKed locally or this property is unavailable at the | ||
| // moment, it is saved even if expired are strictly protected from the removal | ||
| // incl. due to expiration. | ||
| func (x *replicationNode) StoreBinaryObject(cnr cid.ID, bObj []byte) error { | ||
| var obj object.Object | ||
| // TODO(@cthulhu-rider): avoid decoding the object completely | ||
| err := obj.Unmarshal(bObj) | ||
| if err != nil { | ||
| return fmt.Errorf("decode object from binary: %w", err) | ||
| } | ||
|
|
||
| id, ok := obj.ID() | ||
| if !ok { | ||
| return errors.New("missing object ID") | ||
| } | ||
|
|
||
| attrs := obj.Attributes() | ||
| for i := range attrs { | ||
| if attrs[i].Key() != object.AttributeExpirationEpoch { | ||
| continue | ||
| } | ||
|
|
||
| expiresAfter, err := strconv.ParseUint(attrs[i].Value(), 10, 64) | ||
| if err != nil { | ||
| x.log.Debug("failed to decode object expiration attribute, save the object anyway", | ||
| zap.String("value", attrs[i].Value()), zap.Error(err)) | ||
| break | ||
| } | ||
|
|
||
| if x.getCurrentEpoch() <= expiresAfter { | ||
| break | ||
| } | ||
|
|
||
| var addr oid.Address | ||
| addr.SetContainer(cnr) | ||
| addr.SetObject(id) | ||
|
|
||
| locked, err := x.localObjStorage.IsLocked(addr) | ||
| if err != nil { | ||
| x.log.Debug("failed to check whether object is LOCKed, save the object anyway", | ||
| zap.Stringer("address", addr), zap.Error(err)) | ||
| break | ||
| } | ||
|
|
||
| if locked { | ||
| x.log.Debug("object is expired but LOCKed, save the object anyway") | ||
| break | ||
| } | ||
|
|
||
| return errObjectExpired | ||
| } | ||
|
|
||
| var prm engine.PutPrm | ||
| prm.WithObject(&obj) | ||
| _, err = x.localObjStorage.Put(prm) | ||
| if err != nil { | ||
| return fmt.Errorf("save object in the object storage engine: %w", err) | ||
| } | ||
|
|
||
| return nil | ||
| } |
Oops, something went wrong.