achuprov/frostfs-node
1
0
Fork 0
forked from TrueCloudLab/frostfs-node
Code Pull requests Activity

[#217] containersvc: Refactor route passing

Browse source 
Resolve containedctx for routeCtx.

Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
This commit is contained in:
Dmitrii Stepanov 2023-04-05 14:52:13 +03:00 committed by Gitea
parent 6f7b6a8813
commit 279261ace3
10 changed files with 54 additions and 82 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
cmd/frostfs-node/container.go
View file

@ -335,7 +335,7 @@ type remoteLoadAnnounceProvider struct {
deadEndProvider loadcontroller.WriterProvider
}
func (r *remoteLoadAnnounceProvider) InitRemote(srv loadroute.ServerInfo) (loadcontroller.WriterProvider, error) {
func (r *remoteLoadAnnounceProvider) InitRemote(srv loadcontroller.ServerInfo) (loadcontroller.WriterProvider, error) {
if srv == nil {
return r.deadEndProvider, nil
}
@ -366,7 +366,7 @@ type remoteLoadAnnounceWriterProvider struct {
client client.Client
}
func (p *remoteLoadAnnounceWriterProvider) InitWriter(ctx context.Context) (loadcontroller.Writer, error) {
func (p *remoteLoadAnnounceWriterProvider) InitWriter([]loadcontroller.ServerInfo) (loadcontroller.Writer, error) {
return &remoteLoadAnnounceWriter{
client: p.client,
}, nil
@ -536,7 +536,7 @@ func (c *usedSpaceService) ExternalAddresses() []string {
}
func (c *usedSpaceService) AnnounceUsedSpace(ctx context.Context, req *containerV2.AnnounceUsedSpaceRequest) (*containerV2.AnnounceUsedSpaceResponse, error) {
var passedRoute []loadroute.ServerInfo
var passedRoute []loadcontroller.ServerInfo
for hdr := req.GetVerificationHeader(); hdr != nil; hdr = hdr.GetOrigin() {
passedRoute = append(passedRoute, &containerOnlyKeyRemoteServerInfo{
@ -550,7 +550,7 @@ func (c *usedSpaceService) AnnounceUsedSpace(ctx context.Context, req *container
passedRoute = append(passedRoute, c)
w, err := c.loadWriterProvider.InitWriter(loadroute.NewRouteContext(ctx, passedRoute))
w, err := c.loadWriterProvider.InitWriter(passedRoute)
if err != nil {
return nil, fmt.Errorf("could not initialize container's used space writer: %w", err)
}
@ -615,7 +615,7 @@ func (l *loadPlacementBuilder) isNodeFromContainerKey(epoch uint64, cnr cid.ID,
}
func (c *usedSpaceService) processLoadValue(_ context.Context, a containerSDK.SizeEstimation,
route []loadroute.ServerInfo, w loadcontroller.Writer) error {
route []loadcontroller.ServerInfo, w loadcontroller.Writer) error {
fromCnr, err := c.loadPlacementBuilder.isNodeFromContainerKey(a.Epoch(), a.Container(), route[0].PublicKey())
if err != nil {
return fmt.Errorf("could not verify that the sender belongs to the container: %w", err)

8
pkg/services/container/announcement/load/controller/calls.go
View file

@ -62,7 +62,7 @@ func (c *announceContext) announce() {
)
// initialize iterator over locally collected metrics
metricsIterator, err = c.ctrl.prm.LocalMetrics.InitIterator(c.ctx)
metricsIterator, err = c.ctrl.prm.LocalMetrics.InitIterator()
if err != nil {
c.log.Debug("could not initialize iterator over locally collected metrics",
zap.String("error", err.Error()),
@ -72,7 +72,7 @@ func (c *announceContext) announce() {
}
// initialize target of local announcements
targetWriter, err := c.ctrl.prm.LocalAnnouncementTarget.InitWriter(c.ctx)
targetWriter, err := c.ctrl.prm.LocalAnnouncementTarget.InitWriter(nil)
if err != nil {
c.log.Debug("could not initialize announcement accumulator",
zap.String("error", err.Error()),
@ -268,7 +268,7 @@ func (c *stopContext) report() {
)
// initialize iterator over locally accumulated announcements
localIterator, err = c.ctrl.prm.AnnouncementAccumulator.InitIterator(c.ctx)
localIterator, err = c.ctrl.prm.AnnouncementAccumulator.InitIterator()
if err != nil {
c.log.Debug("could not initialize iterator over locally accumulated announcements",
zap.String("error", err.Error()),
@ -278,7 +278,7 @@ func (c *stopContext) report() {
}
// initialize final destination of load estimations
resultWriter, err := c.ctrl.prm.ResultReceiver.InitWriter(c.ctx)
resultWriter, err := c.ctrl.prm.ResultReceiver.InitWriter(nil)
if err != nil {
c.log.Debug("could not initialize result target",
zap.String("error", err.Error()),

4
pkg/services/container/announcement/load/controller/calls_test.go
View file

@ -28,7 +28,7 @@ func newTestStorage() *testAnnouncementStorage {
}
}
func (s *testAnnouncementStorage) InitIterator(context.Context) (loadcontroller.Iterator, error) {
func (s *testAnnouncementStorage) InitIterator() (loadcontroller.Iterator, error) {
if s.i != nil {
return s.i, nil
}
@ -53,7 +53,7 @@ func (s *testAnnouncementStorage) Iterate(f loadcontroller.UsedSpaceFilter, h lo
return nil
}
func (s *testAnnouncementStorage) InitWriter(context.Context) (loadcontroller.Writer, error) {
func (s *testAnnouncementStorage) InitWriter([]loadcontroller.ServerInfo) (loadcontroller.Writer, error) {
if s.w != nil {
return s.w, nil
}

26
pkg/services/container/announcement/load/controller/deps.go
View file

@ -45,7 +45,7 @@ type IteratorProvider interface {
//
// Implementations can have different logic for different
// contexts, so specific ones may document their own behavior.
InitIterator(context.Context) (Iterator, error)
InitIterator() (Iterator, error)
}
// Writer describes the interface for storing container.SizeEstimation values.
@ -80,8 +80,24 @@ type WriterProvider interface {
//
// Initialization problems are reported via error.
// If no error was returned, then the Writer must not be nil.
//
// Implementations can have different logic for different
// contexts, so specific ones may document their own behavior.
InitWriter(context.Context) (Writer, error)
InitWriter(route []ServerInfo) (Writer, error)
}
// ServerInfo describes a set of
// characteristics of a point in a route.
type ServerInfo interface {
// PublicKey returns public key of the node
// from the route in a binary representation.
PublicKey() []byte
// Iterates over network addresses of the node
// in the route. Breaks iterating on true return
// of the handler.
IterateAddresses(func(string) bool)
// Returns number of server's network addresses.
NumberOfAddresses() int
// ExternalAddresses returns external node's addresses.
ExternalAddresses() []string
}

6
pkg/services/container/announcement/load/controller/util.go
View file

@ -1,8 +1,6 @@
package loadcontroller
import (
"context"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container"
)
@ -17,11 +15,11 @@ type storageWrapper struct {
i Iterator
}
func (s storageWrapper) InitIterator(context.Context) (Iterator, error) {
func (s storageWrapper) InitIterator() (Iterator, error) {
return s.i, nil
}
func (s storageWrapper) InitWriter(context.Context) (Writer, error) {
func (s storageWrapper) InitWriter([]ServerInfo) (Writer, error) {
return s.w, nil
}

45
pkg/services/container/announcement/load/route/calls.go
View file

@ -10,26 +10,9 @@ import (
"go.uber.org/zap"
)
// nolint: containedctx
type routeContext struct {
context.Context
passedRoute []ServerInfo
}
// NewRouteContext wraps the main context of value passing with its traversal route.
//
// Passing the result to Router.InitWriter method will allow you to continue this route.
func NewRouteContext(ctx context.Context, passed []ServerInfo) context.Context {
return &routeContext{
Context: ctx,
passedRoute: passed,
}
}
// InitWriter initializes and returns Writer that sends each value to its next route point.
//
// If ctx was created by NewRouteContext, then the traversed route is taken into account,
// If route is present, then it is taken into account,
// and the value will be sent to its continuation. Otherwise, the route will be laid
// from scratch and the value will be sent to its primary point.
//
@ -41,22 +24,14 @@ func NewRouteContext(ctx context.Context, passed []ServerInfo) context.Context {
// runtime and never returns an error.
//
// Always returns nil error.
func (r *Router) InitWriter(ctx context.Context) (loadcontroller.Writer, error) {
var (
routeCtx *routeContext
ok bool
)
if routeCtx, ok = ctx.(*routeContext); !ok {
routeCtx = &routeContext{
Context: ctx,
passedRoute: []ServerInfo{r.localSrvInfo},
}
func (r *Router) InitWriter(route []loadcontroller.ServerInfo) (loadcontroller.Writer, error) {
if len(route) == 0 {
route = []loadcontroller.ServerInfo{r.localSrvInfo}
}
return &loadWriter{
router: r,
ctx: routeCtx,
route: route,
mRoute: make(map[routeKey]*valuesRoute),
mServers: make(map[string]loadcontroller.Writer),
}, nil
@ -69,7 +44,7 @@ type routeKey struct {
}
type valuesRoute struct {
route []ServerInfo
route []loadcontroller.ServerInfo
values []container.SizeEstimation
}
@ -77,7 +52,7 @@ type valuesRoute struct {
type loadWriter struct {
router *Router
ctx *routeContext
route []loadcontroller.ServerInfo
routeMtx sync.RWMutex
mRoute map[routeKey]*valuesRoute
@ -96,11 +71,11 @@ func (w *loadWriter) Put(a container.SizeEstimation) error {
routeValues, ok := w.mRoute[key]
if !ok {
route, err := w.router.routeBuilder.NextStage(a, w.ctx.passedRoute)
route, err := w.router.routeBuilder.NextStage(a, w.route)
if err != nil {
return err
} else if len(route) == 0 {
route = []ServerInfo{nil}
route = []loadcontroller.ServerInfo{nil}
}
routeValues = &valuesRoute{
@ -129,7 +104,7 @@ func (w *loadWriter) Put(a container.SizeEstimation) error {
continue // best effort
}
remoteWriter, err = provider.InitWriter(w.ctx)
remoteWriter, err = provider.InitWriter(w.route)
if err != nil {
w.router.log.Debug("could not initialize writer",
zap.String("error", err.Error()),

23
pkg/services/container/announcement/load/route/deps.go
View file

@ -5,25 +5,6 @@ import (
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container"
)
// ServerInfo describes a set of
// characteristics of a point in a route.
type ServerInfo interface {
// PublicKey returns public key of the node
// from the route in a binary representation.
PublicKey() []byte
// Iterates over network addresses of the node
// in the route. Breaks iterating on true return
// of the handler.
IterateAddresses(func(string) bool)
// Returns number of server's network addresses.
NumberOfAddresses() int
// ExternalAddresses returns external node's addresses.
ExternalAddresses() []string
}
// Builder groups methods to route values in the network.
type Builder interface {
// NextStage must return next group of route points for the value a
@ -36,7 +17,7 @@ type Builder interface {
// in that list (means that point is the last point in one of the route groups),
// returned route must contain nil point that should be interpreted as signal to,
// among sending to other route points, save the announcement in that point.
NextStage(a container.SizeEstimation, passed []ServerInfo) ([]ServerInfo, error)
NextStage(a container.SizeEstimation, passed []loadcontroller.ServerInfo) ([]loadcontroller.ServerInfo, error)
}
// RemoteWriterProvider describes the component
@ -46,5 +27,5 @@ type RemoteWriterProvider interface {
// corresponding to info.
//
// Nil info matches the end of the route.
InitRemote(info ServerInfo) (loadcontroller.WriterProvider, error)
InitRemote(info loadcontroller.ServerInfo) (loadcontroller.WriterProvider, error)
}

6
pkg/services/container/announcement/load/route/placement/calls.go
View file

@ -5,7 +5,7 @@ import (
"fmt"
netmapcore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/netmap"
loadroute "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/container/announcement/load/route"
loadcontroller "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/container/announcement/load/controller"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container"
)
@ -15,7 +15,7 @@ import (
// If passed route has more than one point, then endpoint of the route is reached.
//
// The traversed route is not checked, it is assumed to be correct.
func (b *Builder) NextStage(a container.SizeEstimation, passed []loadroute.ServerInfo) ([]loadroute.ServerInfo, error) {
func (b *Builder) NextStage(a container.SizeEstimation, passed []loadcontroller.ServerInfo) ([]loadcontroller.ServerInfo, error) {
if len(passed) > 1 {
return nil, nil
}
@ -27,7 +27,7 @@ func (b *Builder) NextStage(a container.SizeEstimation, passed []loadroute.Serve
return nil, fmt.Errorf("could not build placement %s: %w", cnr, err)
}
res := make([]loadroute.ServerInfo, 0, len(placement))
res := make([]loadcontroller.ServerInfo, 0, len(placement))
for i := range placement {
if len(placement[i]) == 0 {

5
pkg/services/container/announcement/load/route/router.go
View file

@ -3,6 +3,7 @@ package loadroute
import (
"fmt"
loadcontroller "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/container/announcement/load/controller"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger"
)
@ -15,7 +16,7 @@ type Prm struct {
// Characteristics of the local node's server.
//
// Must not be nil.
LocalServerInfo ServerInfo
LocalServerInfo loadcontroller.ServerInfo
// Component for sending values to a fixed route point.
//
@ -46,7 +47,7 @@ type Router struct {
routeBuilder Builder
localSrvInfo ServerInfo
localSrvInfo loadcontroller.ServerInfo
}
const invalidPrmValFmt = "invalid parameter %s (%T):%v"

3
pkg/services/container/announcement/load/route/util.go
View file

@ -4,6 +4,7 @@ import (
"bytes"
"errors"
loadcontroller "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/container/announcement/load/controller"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container"
)
@ -12,7 +13,7 @@ var errWrongRoute = errors.New("wrong route")
// CheckRoute checks if the route is a route correctly constructed by the builder for value a.
//
// Returns nil if route is correct, otherwise an error clarifying the inconsistency.
func CheckRoute(builder Builder, a container.SizeEstimation, route []ServerInfo) error {
func CheckRoute(builder Builder, a container.SizeEstimation, route []loadcontroller.ServerInfo) error {
for i := 1; i < len(route); i++ {
servers, err := builder.NextStage(a, route[:i])
if err != nil {