Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
node/meta: catch meta notifications from chain
Store caught information in MPT structures, one structure by one container. Bbolt KV database is used as backend for MPT tries. Closes #3070. Signed-off-by: Pavel Karpy <[email protected]>
- Loading branch information
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,17 @@ | ||
| package metaconfig | ||
|
|
||
| import ( | ||
| "github.com/nspcc-dev/neofs-node/cmd/neofs-node/config" | ||
| ) | ||
|
|
||
| const ( | ||
| subsection = "metadata" | ||
| ) | ||
|
|
||
| // Path returns the value of "path" config parameter | ||
| // from "metadata" section. | ||
| // | ||
| // Returns empty string if the value is missing or invalid. | ||
| func Path(c *config.Config) string { | ||
| return config.StringSafe(c.Sub(subsection), "path") | ||
| } |
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,29 @@ | ||
| package metaconfig_test | ||
|
|
||
| import ( | ||
| "testing" | ||
|
|
||
| "github.com/nspcc-dev/neofs-node/cmd/neofs-node/config" | ||
| metaconfig "github.com/nspcc-dev/neofs-node/cmd/neofs-node/config/meta" | ||
| configtest "github.com/nspcc-dev/neofs-node/cmd/neofs-node/config/test" | ||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| func TestLoggerSection_Level(t *testing.T) { | ||
| t.Run("defaults", func(t *testing.T) { | ||
| emptyConfig := configtest.EmptyConfig() | ||
| require.Equal(t, "", metaconfig.Path(emptyConfig)) | ||
| }) | ||
|
|
||
| const path = "../../../../config/example/node" | ||
|
|
||
| var fileConfigTest = func(c *config.Config) { | ||
| require.Equal(t, "path/to/meta", metaconfig.Path(c)) | ||
| } | ||
|
|
||
| configtest.ForEachFileType(path, fileConfigTest) | ||
|
|
||
| t.Run("ENV", func(t *testing.T) { | ||
| configtest.ForEnvFileType(path, fileConfigTest) | ||
| }) | ||
| } |
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,125 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "context" | ||
| "fmt" | ||
| "slices" | ||
| "sync" | ||
|
|
||
| "github.com/nspcc-dev/neofs-node/pkg/core/container" | ||
| "github.com/nspcc-dev/neofs-node/pkg/core/netmap" | ||
| cntClient "github.com/nspcc-dev/neofs-node/pkg/morph/client/container" | ||
| "github.com/nspcc-dev/neofs-node/pkg/services/meta" | ||
| cid "github.com/nspcc-dev/neofs-sdk-go/container/id" | ||
| netmapsdk "github.com/nspcc-dev/neofs-sdk-go/netmap" | ||
| "go.uber.org/zap" | ||
| "golang.org/x/sync/errgroup" | ||
| ) | ||
|
|
||
| func initMeta(c *cfg) { | ||
| if c.cfgMorph.client == nil { | ||
| initMorphComponents(c) | ||
| } | ||
|
|
||
| c.cfgMeta.cLister = &containerListener{ | ||
| key: c.binPublicKey, | ||
| cnrClient: c.basics.cCli, | ||
| containers: c.cfgObject.cnrSource, | ||
| network: c.basics.netMapSource, | ||
| } | ||
|
|
||
| m, err := meta.New(c.log.With(zap.String("service", "meta data")), c.cfgMeta.cLister, c.basics.wsCli, c.basics.containerSH, c.basics.netmapSH, c.applicationConfiguration.metadata.path) | ||
| fatalOnErr(err) | ||
|
|
||
| c.workers = append(c.workers, newWorkerFromFunc(func(ctx context.Context) { | ||
| err = m.Run(ctx) | ||
| if err != nil { | ||
| c.internalErr <- fmt.Errorf("meta data service error: %w", err) | ||
| } | ||
| })) | ||
| } | ||
|
|
||
| type containerListener struct { | ||
| key []byte | ||
|
|
||
| cnrClient *cntClient.Client | ||
| containers container.Source | ||
| network netmap.Source | ||
|
|
||
| m sync.RWMutex | ||
| prevCnrs []cid.ID | ||
| prevNetMap *netmapsdk.NetMap | ||
| prevRes map[cid.ID]struct{} | ||
| } | ||
|
|
||
| func (c *containerListener) List() (map[cid.ID]struct{}, error) { | ||
| actualContainers, err := c.cnrClient.List(nil) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("read containers: %w", err) | ||
| } | ||
| curEpoch, err := c.network.Epoch() | ||
| if err != nil { | ||
| return nil, fmt.Errorf("read current NeoFS epoch: %w", err) | ||
| } | ||
| networkMap, err := c.network.GetNetMapByEpoch(curEpoch) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("read network map at the current epoch #%d: %w", curEpoch, err) | ||
| } | ||
|
|
||
| c.m.RLock() | ||
| cnrsSame := slices.EqualFunc(c.prevCnrs, actualContainers, func(cID1, cID2 cid.ID) bool { | ||
| return bytes.Equal(cID1[:], cID2[:]) | ||
| }) | ||
| netmapSame := slices.EqualFunc(c.prevNetMap.Nodes(), networkMap.Nodes(), func(n1 netmapsdk.NodeInfo, n2 netmapsdk.NodeInfo) bool { | ||
| return bytes.Equal(n1.PublicKey(), n2.PublicKey()) | ||
| }) | ||
| if cnrsSame && netmapSame && c.prevRes != nil { | ||
| c.m.RUnlock() | ||
| return c.prevRes, nil | ||
| } | ||
| c.m.RUnlock() | ||
|
|
||
| var locM sync.Mutex | ||
| res := make(map[cid.ID]struct{}) | ||
| var wg errgroup.Group | ||
| for _, cID := range actualContainers { | ||
| wg.Go(func() error { | ||
| cnr, err := c.containers.Get(cID) | ||
| if err != nil { | ||
| return fmt.Errorf("read %s container: %w", cID, err) | ||
| } | ||
|
|
||
| nodeSets, err := networkMap.ContainerNodes(cnr.Value.PlacementPolicy(), cID) | ||
| if err != nil { | ||
| return fmt.Errorf("apply container storage policy to %s container: %w", cID, err) | ||
| } | ||
|
|
||
| for _, nodeSet := range nodeSets { | ||
| for _, node := range nodeSet { | ||
| if bytes.Equal(node.PublicKey(), c.key) { | ||
| locM.Lock() | ||
| res[cID] = struct{}{} | ||
| locM.Unlock() | ||
| return nil | ||
| } | ||
| } | ||
| } | ||
|
|
||
| return nil | ||
| }) | ||
| } | ||
|
|
||
| err = wg.Wait() | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| c.m.Lock() | ||
| c.prevCnrs = actualContainers | ||
| c.prevNetMap = networkMap | ||
| c.prevRes = res | ||
| c.m.Unlock() | ||
|
|
||
| return res, nil | ||
| } | ||
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