forked from TrueCloudLab/frostfs-node
[#1902] tree: Allow synchronize all the container trees
Add `SynchronizeAllTrees` method of the Tree service. It allows fetching tree IDs and sync all of them. Share common logic b/w the new method and the `SynchronizeTree`. Signed-off-by: Pavel Karpy <carpawell@nspcc.ru>
This commit is contained in:
parent
1805c6606d
commit
1766ca2039
3 changed files with 103 additions and 7 deletions
|
@ -7,9 +7,19 @@ import (
|
||||||
controlconfig "github.com/nspcc-dev/neofs-node/cmd/neofs-node/config/control"
|
controlconfig "github.com/nspcc-dev/neofs-node/cmd/neofs-node/config/control"
|
||||||
"github.com/nspcc-dev/neofs-node/pkg/services/control"
|
"github.com/nspcc-dev/neofs-node/pkg/services/control"
|
||||||
controlSvc "github.com/nspcc-dev/neofs-node/pkg/services/control/server"
|
controlSvc "github.com/nspcc-dev/neofs-node/pkg/services/control/server"
|
||||||
|
"github.com/nspcc-dev/neofs-node/pkg/services/tree"
|
||||||
|
cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
|
||||||
"google.golang.org/grpc"
|
"google.golang.org/grpc"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
type treeSynchronizer struct {
|
||||||
|
treeSvc *tree.Service
|
||||||
|
}
|
||||||
|
|
||||||
|
func (t treeSynchronizer) Synchronize(ctx context.Context, cnr cid.ID, treeID string) error {
|
||||||
|
return t.treeSvc.SynchronizeTree(ctx, cnr, treeID)
|
||||||
|
}
|
||||||
|
|
||||||
func initControlService(c *cfg) {
|
func initControlService(c *cfg) {
|
||||||
endpoint := controlconfig.GRPC(c.appCfg).Endpoint()
|
endpoint := controlconfig.GRPC(c.appCfg).Endpoint()
|
||||||
if endpoint == controlconfig.GRPCEndpointDefault {
|
if endpoint == controlconfig.GRPCEndpointDefault {
|
||||||
|
@ -34,7 +44,9 @@ func initControlService(c *cfg) {
|
||||||
controlSvc.WithReplicator(c.replicator),
|
controlSvc.WithReplicator(c.replicator),
|
||||||
controlSvc.WithNodeState(c),
|
controlSvc.WithNodeState(c),
|
||||||
controlSvc.WithLocalStorage(c.cfgObject.cfgLocalStorage.localStorage),
|
controlSvc.WithLocalStorage(c.cfgObject.cfgLocalStorage.localStorage),
|
||||||
controlSvc.WithTreeService(c.treeService),
|
controlSvc.WithTreeService(treeSynchronizer{
|
||||||
|
c.treeService,
|
||||||
|
}),
|
||||||
)
|
)
|
||||||
|
|
||||||
lis, err := net.Listen("tcp", endpoint)
|
lis, err := net.Listen("tcp", endpoint)
|
||||||
|
|
|
@ -80,13 +80,13 @@ func syncTrees(ctx context.Context, treeSvc *tree.Service, cnrCli *containerClie
|
||||||
wellKnownTrees := [...]string{"version", "system"}
|
wellKnownTrees := [...]string{"version", "system"}
|
||||||
|
|
||||||
for _, id := range ids {
|
for _, id := range ids {
|
||||||
for _, tID := range wellKnownTrees {
|
for i := range wellKnownTrees {
|
||||||
err = treeSvc.Synchronize(ctx, id, tID)
|
err = treeSvc.SynchronizeTree(ctx, id, wellKnownTrees[i])
|
||||||
if err != nil && !errors.Is(err, tree.ErrNotInContainer) {
|
if err != nil && !errors.Is(err, tree.ErrNotInContainer) {
|
||||||
log.Warn(
|
log.Warn(
|
||||||
"tree synchronization failed",
|
"tree synchronization failed",
|
||||||
zap.Stringer("cid", id),
|
zap.Stringer("cid", id),
|
||||||
zap.String("tree_id", tID),
|
zap.String("tree_id", wellKnownTrees[i]),
|
||||||
zap.Error(err),
|
zap.Error(err),
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
|
@ -10,6 +10,8 @@ import (
|
||||||
"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/pilorama"
|
"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/pilorama"
|
||||||
"github.com/nspcc-dev/neofs-node/pkg/network"
|
"github.com/nspcc-dev/neofs-node/pkg/network"
|
||||||
cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
|
cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
|
||||||
|
netmapSDK "github.com/nspcc-dev/neofs-sdk-go/netmap"
|
||||||
|
"go.uber.org/zap"
|
||||||
"google.golang.org/grpc"
|
"google.golang.org/grpc"
|
||||||
"google.golang.org/grpc/credentials/insecure"
|
"google.golang.org/grpc/credentials/insecure"
|
||||||
)
|
)
|
||||||
|
@ -17,9 +19,12 @@ import (
|
||||||
// ErrNotInContainer is returned when operation could not be performed
|
// ErrNotInContainer is returned when operation could not be performed
|
||||||
// because the node is not included in the container.
|
// because the node is not included in the container.
|
||||||
var ErrNotInContainer = errors.New("node is not in container")
|
var ErrNotInContainer = errors.New("node is not in container")
|
||||||
|
var errNoOtherNodes = errors.New("no nodes to fetch trees from")
|
||||||
|
|
||||||
// Synchronize tries to synchronize log starting from the last stored height.
|
// SynchronizeAllTrees synchronizes all the trees of the container. It fetches
|
||||||
func (s *Service) Synchronize(ctx context.Context, cid cid.ID, treeID string) error {
|
// tree IDs from the other container nodes. Returns ErrNotInContainer if the node
|
||||||
|
// is not included in the container.
|
||||||
|
func (s *Service) SynchronizeAllTrees(ctx context.Context, cid cid.ID) error {
|
||||||
nodes, pos, err := s.getContainerNodes(cid)
|
nodes, pos, err := s.getContainerNodes(cid)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("can't get container nodes: %w", err)
|
return fmt.Errorf("can't get container nodes: %w", err)
|
||||||
|
@ -34,7 +39,86 @@ func (s *Service) Synchronize(ctx context.Context, cid cid.ID, treeID string) er
|
||||||
d.Position = pos
|
d.Position = pos
|
||||||
d.Size = len(nodes)
|
d.Size = len(nodes)
|
||||||
|
|
||||||
lm, err := s.forest.TreeGetOpLog(cid, treeID, 0)
|
nodes = append(nodes[:pos], nodes[pos+1:]...) // exclude that node
|
||||||
|
if len(nodes) == 0 {
|
||||||
|
return errNoOtherNodes
|
||||||
|
}
|
||||||
|
|
||||||
|
rawCID := make([]byte, sha256.Size)
|
||||||
|
cid.Encode(rawCID)
|
||||||
|
|
||||||
|
req := &TreeListRequest{
|
||||||
|
Body: &TreeListRequest_Body{
|
||||||
|
ContainerId: rawCID,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
err = SignMessage(req, s.key)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("could not sign request: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
var resp *TreeListResponse
|
||||||
|
var treesToSync []string
|
||||||
|
var outErr error
|
||||||
|
|
||||||
|
err = s.forEachNode(ctx, nodes, func(c TreeServiceClient) bool {
|
||||||
|
resp, outErr = c.TreeList(ctx, req)
|
||||||
|
if outErr != nil {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
treesToSync = resp.GetBody().GetIds()
|
||||||
|
|
||||||
|
return true
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
outErr = err
|
||||||
|
}
|
||||||
|
|
||||||
|
if outErr != nil {
|
||||||
|
return fmt.Errorf("could not fetch tree ID list: %w", outErr)
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tid := range treesToSync {
|
||||||
|
err = s.synchronizeTree(ctx, d, tid, nodes)
|
||||||
|
if err != nil {
|
||||||
|
s.log.Error("could not sync tree",
|
||||||
|
zap.Stringer("cid", cid),
|
||||||
|
zap.String("treeID", tid))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// SynchronizeTree tries to synchronize log starting from the last stored height.
|
||||||
|
func (s *Service) SynchronizeTree(ctx context.Context, cid cid.ID, treeID string) error {
|
||||||
|
nodes, pos, err := s.getContainerNodes(cid)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("can't get container nodes: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if pos < 0 {
|
||||||
|
return ErrNotInContainer
|
||||||
|
}
|
||||||
|
|
||||||
|
var d pilorama.CIDDescriptor
|
||||||
|
d.CID = cid
|
||||||
|
d.Position = pos
|
||||||
|
d.Size = len(nodes)
|
||||||
|
|
||||||
|
nodes = append(nodes[:pos], nodes[pos+1:]...) // exclude that node
|
||||||
|
if len(nodes) == 0 {
|
||||||
|
return errNoOtherNodes
|
||||||
|
}
|
||||||
|
|
||||||
|
return s.synchronizeTree(ctx, d, treeID, nodes)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Service) synchronizeTree(ctx context.Context, d pilorama.CIDDescriptor,
|
||||||
|
treeID string, nodes []netmapSDK.NodeInfo) error {
|
||||||
|
lm, err := s.forest.TreeGetOpLog(d.CID, treeID, 0)
|
||||||
if err != nil && !errors.Is(err, pilorama.ErrTreeNotFound) {
|
if err != nil && !errors.Is(err, pilorama.ErrTreeNotFound) {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in a new issue