-
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.
node/object: Serve new 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. Requests are served similar to `ObjectService.Put` ones with TTL=1 (local only). Refs #2317. Signed-off-by: Leonard Lyubich <leonard@morphbits.io>
- Loading branch information
1 parent
2b069f6
commit 2fa7832
Showing
10 changed files
with
1,274 additions
and
16 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,127 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "fmt" | ||
|
|
||
| putsvc "github.com/nspcc-dev/neofs-node/pkg/services/object/put" | ||
| 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" | ||
| "go.uber.org/zap" | ||
| ) | ||
|
|
||
| // replicationNode checks object storage policy compliance against NeoFS network | ||
| // maps. | ||
| type replicationNode struct { | ||
| log *zap.Logger | ||
|
|
||
| putObjectService *putsvc.Service | ||
|
|
||
| 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, | ||
| putObjectService *putsvc.Service, | ||
| 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, | ||
| putObjectService: putObjectService, | ||
| 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 | ||
| } | ||
|
|
||
| // StoreObject processes object same way as `ObjectService.Put` RPC with TTL=1. | ||
| func (x *replicationNode) StoreObject(cnr cid.ID, obj object.Object) error { | ||
| return x.putObjectService.ValidateAndStoreObjectLocally(cnr, obj) | ||
| } |
Oops, something went wrong.